fn replicate_state(
        obj: &SharedSkel,
        verified_receiver: &streamer::BlobReceiver,
        blob_recycler: &packet::BlobRecycler,
    ) -> Result<()> {
        let timer = Duration::new(1, 0);
        let blobs = verified_receiver.recv_timeout(timer)?;
        trace!("replicating blobs {}", blobs.len());
        for msgs in &blobs {
            let blob = msgs.read().unwrap();
            let entries: Vec<Entry> = deserialize(&blob.data()[..blob.meta.size]).unwrap();
            let acc = obj.acc.lock().unwrap();
            for entry in entries {
                acc.register_entry_id(&entry.id);
                for result in acc.process_verified_events(entry.events) {
                    result?;
                }
            }
            //TODO respond back to leader with hash of the state
        }
        for blob in blobs {
            blob_recycler.recycle(blob);
        }
        Ok(())
    }
/// Process verified blobs, already in order
/// Respond with a signed hash of the state
Process verified blobs, already in order
Respond with a signed hash of the state
Process verified blobs already in order Respond with a signed hash of the state
fn replicate_state(
        obj: &SharedSkel,
        verified_receiver: &streamer::BlobReceiver,
        blob_recycler: &packet::BlobRecycler,
    ) -> Result<()> {
        let timer = Duration::new(1, 0);
        let blobs = verified_receiver.recv_timeout(timer)?;
        trace!("replicating blobs {}", blobs.len());
        for msgs in &blobs {
            let blob = msgs.read().unwrap();
            let entries: Vec<Entry> = deserialize(&blob.data()[..blob.meta.size]).unwrap();
            let acc = obj.acc.lock().unwrap();
            for entry in entries {
                acc.register_entry_id(&entry.id);
                for result in acc.process_verified_events(entry.events) {
                    result?;
                }
            }
            //TODO respond back to leader with hash of the state
        }
        for blob in blobs {
            blob_recycler.recycle(blob);
        }
        Ok(())
    }
fn replicate_state ( obj : & SharedSkel , verified_receiver : & streamer :: BlobReceiver , blob_recycler : & packet :: BlobRecycler , ) -> Result < ( ) > { let timer = Duration :: new ( 1 , 0 ) ; let blobs = verified_receiver . recv_timeout ( timer ) ? ; trace ! ( " " blobs len ( ) ) ; for msgs in & blobs { let blob = msgs . read ( ) . unwrap ( ) ; let entries : Vec < Entry > = deserialize ( & blob . data ( ) [ .. blob . meta . size ] ) . unwrap ( ) ; let acc = obj . acc . lock ( ) . unwrap ( ) ; for entry in entries { acc . register_entry_id ( & entry . id ) ; for result in acc . process_verified_events ( entry . events ) { result ? ; } } //TODO respond back to leader with hash of the state } for blob in blobs { blob_recycler . recycle ( blob ) ; } Ok ( ( ) ) }
Process verified blobs, already in order
Respond with a signed hash of the state
Process verified blobs already in order Respond with a signed hash of the state
obj
&SharedSkel
verified_receiver
&streamer::BlobReceiver
blob_recycler
&packet::BlobRecycler
obj
&SharedSkel
verified_receiver
&streamer::BlobReceiver
blob_recycler
&packet::BlobRecycler