Refactor such that genesis can use entry_writer
This commit is contained in:
parent
1c35d59f26
commit
b60802ddff
|
@ -5,10 +5,12 @@ extern crate serde_json;
|
||||||
extern crate solana;
|
extern crate solana;
|
||||||
|
|
||||||
use atty::{is, Stream};
|
use atty::{is, Stream};
|
||||||
|
use solana::entry_writer::EntryWriter;
|
||||||
use solana::mint::Mint;
|
use solana::mint::Mint;
|
||||||
use std::error;
|
use std::error;
|
||||||
use std::io::{stdin, stdout, Read, Write};
|
use std::io::{stdin, stdout, Read};
|
||||||
use std::process::exit;
|
use std::process::exit;
|
||||||
|
use std::sync::Mutex;
|
||||||
|
|
||||||
fn main() -> Result<(), Box<error::Error>> {
|
fn main() -> Result<(), Box<error::Error>> {
|
||||||
if is(Stream::Stdin) {
|
if is(Stream::Stdin) {
|
||||||
|
@ -24,9 +26,7 @@ fn main() -> Result<(), Box<error::Error>> {
|
||||||
}
|
}
|
||||||
|
|
||||||
let mint: Mint = serde_json::from_str(&buffer)?;
|
let mint: Mint = serde_json::from_str(&buffer)?;
|
||||||
let mut writer = stdout();
|
let writer = Mutex::new(stdout());
|
||||||
for x in mint.create_entries() {
|
EntryWriter::write_entries(&writer, &mint.create_entries())?;
|
||||||
writeln!(writer, "{}", serde_json::to_string(&x)?)?;
|
|
||||||
}
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
|
@ -25,6 +25,18 @@ impl<'a> EntryWriter<'a> {
|
||||||
EntryWriter { bank }
|
EntryWriter { bank }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn write_entry<W: Write>(writer: &Mutex<W>, entry: &Entry) -> io::Result<()> {
|
||||||
|
let serialized = serde_json::to_string(&entry).unwrap();
|
||||||
|
writeln!(writer.lock().unwrap(), "{}", serialized)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn write_entries<W: Write>(writer: &Mutex<W>, entries: &[Entry]) -> io::Result<()> {
|
||||||
|
for entry in entries {
|
||||||
|
Self::write_entry(writer, entry)?;
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
fn write_and_register_entry<W: Write>(
|
fn write_and_register_entry<W: Write>(
|
||||||
&self,
|
&self,
|
||||||
writer: &Mutex<W>,
|
writer: &Mutex<W>,
|
||||||
|
@ -34,30 +46,26 @@ impl<'a> EntryWriter<'a> {
|
||||||
if !entry.has_more {
|
if !entry.has_more {
|
||||||
self.bank.register_entry_id(&entry.id);
|
self.bank.register_entry_id(&entry.id);
|
||||||
}
|
}
|
||||||
writeln!(
|
Self::write_entry(&writer, entry)
|
||||||
writer
|
|
||||||
.lock()
|
|
||||||
.expect("'writer' lock in fn fn write_and_register_entry"),
|
|
||||||
"{}",
|
|
||||||
serde_json::to_string(&entry)
|
|
||||||
.expect("'entry' to_strong in fn write_and_register_entry")
|
|
||||||
)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn write_entries<W: Write>(
|
fn write_and_register_entries<W: Write>(
|
||||||
&self,
|
&self,
|
||||||
writer: &Mutex<W>,
|
writer: &Mutex<W>,
|
||||||
entry_receiver: &Receiver<Entry>,
|
entries: &[Entry],
|
||||||
) -> Result<Vec<Entry>> {
|
) -> io::Result<()> {
|
||||||
//TODO implement a serialize for channel that does this without allocations
|
for entry in entries {
|
||||||
|
self.write_and_register_entry(writer, &entry)?;
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn recv_entries(entry_receiver: &Receiver<Entry>) -> Result<Vec<Entry>> {
|
||||||
let entry = entry_receiver.recv_timeout(Duration::new(1, 0))?;
|
let entry = entry_receiver.recv_timeout(Duration::new(1, 0))?;
|
||||||
let mut entries = vec![entry];
|
let mut entries = vec![entry];
|
||||||
while let Ok(entry) = entry_receiver.try_recv() {
|
while let Ok(entry) = entry_receiver.try_recv() {
|
||||||
entries.push(entry);
|
entries.push(entry);
|
||||||
}
|
}
|
||||||
for entry in &entries {
|
|
||||||
self.write_and_register_entry(writer, &entry)?;
|
|
||||||
}
|
|
||||||
Ok(entries)
|
Ok(entries)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -70,7 +78,8 @@ impl<'a> EntryWriter<'a> {
|
||||||
writer: &Mutex<W>,
|
writer: &Mutex<W>,
|
||||||
entry_receiver: &Receiver<Entry>,
|
entry_receiver: &Receiver<Entry>,
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let entries = self.write_entries(writer, entry_receiver)?;
|
let entries = Self::recv_entries(entry_receiver)?;
|
||||||
|
self.write_and_register_entries(writer, &entries)?;
|
||||||
trace!("New blobs? {}", entries.len());
|
trace!("New blobs? {}", entries.len());
|
||||||
let mut blobs = VecDeque::new();
|
let mut blobs = VecDeque::new();
|
||||||
entries.to_blobs(blob_recycler, &mut blobs);
|
entries.to_blobs(blob_recycler, &mut blobs);
|
||||||
|
@ -84,7 +93,8 @@ impl<'a> EntryWriter<'a> {
|
||||||
/// Process any Entry items that have been published by the Historian.
|
/// Process any Entry items that have been published by the Historian.
|
||||||
/// continuosly broadcast blobs of entries out
|
/// continuosly broadcast blobs of entries out
|
||||||
pub fn drain_entries(&self, entry_receiver: &Receiver<Entry>) -> Result<()> {
|
pub fn drain_entries(&self, entry_receiver: &Receiver<Entry>) -> Result<()> {
|
||||||
self.write_entries(&Arc::new(Mutex::new(sink())), entry_receiver)?;
|
let entries = Self::recv_entries(entry_receiver)?;
|
||||||
|
self.write_and_register_entries(&Arc::new(Mutex::new(sink())), &entries)?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue