cleanup

src/replicate_stage.rs

@@ -1,10 +1,13 @@
 //! The `replicate_stage` replicates transactions broadcast by the leader.
 
+use bank::Bank;
+use ledger;
 use packet;
-use request_replicator::RequestReplicator;
+use result::Result;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::thread::{spawn, JoinHandle};
+use std::time::Duration;
 use streamer;
 
 pub struct ReplicateStage {
@@ -12,14 +15,34 @@ pub struct ReplicateStage {
 }
 
 impl ReplicateStage {
+    /// Process verified blobs, already in order
+    fn replicate_requests(
+        bank: &Arc<Bank>,
+        verified_receiver: &streamer::BlobReceiver,
+        blob_recycler: &packet::BlobRecycler,
+    ) -> Result<()> {
+        let timer = Duration::new(1, 0);
+        let blobs = verified_receiver.recv_timeout(timer)?;
+        let entries = ledger::reconstruct_entries_from_blobs(&blobs);
+        let res = bank.process_verified_entries(entries);
+        if res.is_err() {
+            error!("process_verified_entries {} {:?}", blobs.len(), res);
+        }
+        res?;
+        for blob in blobs {
+            blob_recycler.recycle(blob);
+        }
+        Ok(())
+    }
+
     pub fn new(
-        request_replicator: RequestReplicator,
+        bank: Arc<Bank>,
         exit: Arc<AtomicBool>,
         window_receiver: streamer::BlobReceiver,
         blob_recycler: packet::BlobRecycler,
     ) -> Self {
         let thread_hdl = spawn(move || loop {
-            let e = request_replicator.replicate_requests(&window_receiver, &blob_recycler);
+            let e = Self::replicate_requests(&bank, &window_receiver, &blob_recycler);
             if e.is_err() && exit.load(Ordering::Relaxed) {
                 break;
             }