progress

src/crdt.rs

@@ -32,7 +32,7 @@ use packet::SharedBlob;
 /// Structure to be replicated by the network
 #[derive(Serialize, Deserialize, Clone)]
 pub struct ReplicatedData {
-    id: PublicKey,
+    pub id: PublicKey,
     sig: Signature,
     /// should always be increasing
     version: u64,

src/lib.rs

@@ -17,7 +17,7 @@ pub mod plan;
 pub mod recorder;
 pub mod result;
 pub mod signature;
-//pub mod streamer;
+pub mod streamer;
 pub mod transaction;
 extern crate bincode;
 extern crate byteorder;

src/streamer.rs

@@ -111,7 +111,7 @@ pub fn blob_receiver(
 
 fn recv_window(
     window: &mut Vec<Option<SharedBlob>>,
-    subs: &Arc<RwLock<Crdt>>,
+    crdt: &Arc<RwLock<Crdt>>,
     recycler: &BlobRecycler,
     consumed: &mut usize,
     r: &BlobReceiver,
@@ -127,7 +127,6 @@ fn recv_window(
     {
         //retransmit all leader blocks
         let mut retransmitq = VecDeque::new();
-        let rsubs = subs.read().unwrap();
         for b in &dq {
             let p = b.read().unwrap();
             //TODO this check isn't safe against adverserial packets
@@ -137,9 +136,9 @@ fn recv_window(
                 p.get_index().unwrap(),
                 p.get_id().unwrap(),
                 p.meta.addr(),
-                rsubs.leader.addr
+                leader_id
             );
-            if p.get_id() == Ok(leader_id) {
+            if p.get_id().unwrap() == leader_id {
                 //TODO
                 //need to copy the retransmited blob
                 //otherwise we get into races with which thread