gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

comments

Browse Source
This commit is contained in:
Anatoly Yakovenko
2018-04-17 11:05:15 -07:00
parent fb55ab8c33
commit 374c9921fd
2 changed files with 22 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/streamer.rs
View File

@@ -222,6 +222,11 @@ fn retransmit(
Ok(())
}
//service to retransmit messages from the leader to layer 1 nodes
//see subscriber.rs for network layer definitions
//window receives blobs from the network
//for any blobs that originated from the leader, we broadcast
//to the rest of the network
pub fn retransmitter(
sock: UdpSocket,
exit: Arc<AtomicBool>,
@@ -485,18 +490,18 @@ mod test {
)));
let n3 = Node::new([0; 8], 1, read.local_addr().unwrap());
subs.write().unwrap().insert(&[n3]);
let (s_cast, r_cast) = channel();
let re = BlobRecycler::default();
let (s_retransmit, r_retransmit) = channel();
let blob_recycler = BlobRecycler::default();
let saddr = send.local_addr().unwrap();
let t_retransmit = retransmitter(send, exit.clone(), subs, re.clone(), r_cast);
let t_retransmit = retransmitter(send, exit.clone(), subs, blob_recycler.clone(), r_retransmit);
let mut bq = VecDeque::new();
let b = re.allocate();
let b = blob_recycler.allocate();
b.write().unwrap().meta.size = 10;
bq.push_back(b);
s_cast.send(bq).unwrap();
let (s_recv, r_recv) = channel();
let t_receiver = blob_receiver(exit.clone(), re.clone(), read, s_recv).unwrap();
let mut oq = r_recv.recv().unwrap();
s_retransmit.send(bq).unwrap();
let (s_blob_receiver, r_blob_receiver) = channel();
let t_receiver = blob_receiver(exit.clone(), blob_recycler.clone(), read, s_blob_receiver).unwrap();
let mut oq = r_blob_receiver.recv().unwrap();
assert_eq!(oq.len(), 1);
let o = oq.pop_front().unwrap();
let ro = o.read().unwrap();