//retransmit from leader to replicate target
retransmit from leader to replicate target
retransmit from leader to replicate target
pub fn retransmit() {
        logger::setup();
        trace!("retransmit test start");
        let exit = Arc::new(AtomicBool::new(false));
        let (crdt_leader, sock_gossip_leader, _, sock_leader) = test_node();
        let (crdt_target, sock_gossip_target, sock_replicate_target, _) = test_node();
        let leader_data = crdt_leader.read().unwrap().my_data().clone();
        crdt_leader.write().unwrap().insert(leader_data.clone());
        crdt_leader.write().unwrap().set_leader(leader_data.id);
        let t_crdt_leader_g = Crdt::gossip(crdt_leader.clone(), exit.clone());
        let t_crdt_leader_l = Crdt::listen(crdt_leader.clone(), sock_gossip_leader, exit.clone());
        crdt_target.write().unwrap().insert(leader_data.clone());
        crdt_target.write().unwrap().set_leader(leader_data.id);
        let t_crdt_target_g = Crdt::gossip(crdt_target.clone(), exit.clone());
        let t_crdt_target_l = Crdt::listen(crdt_target.clone(), sock_gossip_target, exit.clone());
        //leader retransmitter
        let (s_retransmit, r_retransmit) = channel();
        let blob_recycler = BlobRecycler::default();
        let saddr = sock_leader.local_addr().unwrap();
        let t_retransmit = retransmitter(
            sock_leader,
            exit.clone(),
            crdt_leader.clone(),
            blob_recycler.clone(),
            r_retransmit,
        );
        //target receiver
        let (s_blob_receiver, r_blob_receiver) = channel();
        let t_receiver = blob_receiver(
            exit.clone(),
            blob_recycler.clone(),
            sock_replicate_target,
            s_blob_receiver,
        ).unwrap();
        for _ in 0..10 {
            let done = crdt_target.read().unwrap().update_index == 2
                && crdt_leader.read().unwrap().update_index == 2;
            if done {
                break;
            }
            let timer = Duration::new(1, 0);
            sleep(timer);
        }
        //send the data through
        let mut bq = VecDeque::new();
        let b = blob_recycler.allocate();
        b.write().unwrap().meta.size = 10;
        bq.push_back(b);
        s_retransmit.send(bq).unwrap();
        let timer = Duration::new(5, 0);
        trace!("Waiting for timeout");
        let mut oq = r_blob_receiver.recv_timeout(timer).unwrap();
        assert_eq!(oq.len(), 1);
        let o = oq.pop_front().unwrap();
        let ro = o.read().unwrap();
        assert_eq!(ro.meta.size, 10);
        assert_eq!(ro.meta.addr(), saddr);
        exit.store(true, Ordering::Relaxed);
        let threads = vec![
            t_receiver,
            t_retransmit,
            t_crdt_target_g,
            t_crdt_target_l,
            t_crdt_leader_g,
            t_crdt_leader_l,
        ];
        for t in threads {
            t.join().unwrap();
        }
    }
pub fn retransmit ( ) { logger :: setup ( ) ; trace ! ( " " ) ; let exit = Arc :: new ( AtomicBool :: new ( false ) ) ; let ( crdt_leader , sock_gossip_leader , _ , sock_leader ) = test_node ( ) ; let ( crdt_target , sock_gossip_target , sock_replicate_target , _ ) = test_node ( ) ; let leader_data = crdt_leader . read ( ) . unwrap ( ) . my_data ( ) . clone ( ) ; crdt_leader . write ( ) . unwrap ( ) . insert ( leader_data . clone ( ) ) ; crdt_leader . write ( ) . unwrap ( ) . set_leader ( leader_data . id ) ; let t_crdt_leader_g = Crdt :: gossip ( crdt_leader . clone ( ) , exit . clone ( ) ) ; let t_crdt_leader_l = Crdt :: listen ( crdt_leader . clone ( ) , sock_gossip_leader , exit . clone ( ) ) ; crdt_target . write ( ) . unwrap ( ) . insert ( leader_data . clone ( ) ) ; crdt_target . write ( ) . unwrap ( ) . set_leader ( leader_data . id ) ; let t_crdt_target_g = Crdt :: gossip ( crdt_target . clone ( ) , exit . clone ( ) ) ; let t_crdt_target_l = Crdt :: listen ( crdt_target . clone ( ) , sock_gossip_target , exit . clone ( ) ) ; //leader retransmitter let ( s_retransmit , r_retransmit ) = channel ( ) ; let blob_recycler = BlobRecycler :: default ( ) ; let saddr = sock_leader . local_addr ( ) . unwrap ( ) ; let t_retransmit = retransmitter ( sock_leader , exit . clone ( ) , crdt_leader . clone ( ) , blob_recycler . clone ( ) , r_retransmit , ) ; //target receiver let ( s_blob_receiver , r_blob_receiver ) = channel ( ) ; let t_receiver = blob_receiver ( exit . clone ( ) , blob_recycler . clone ( ) , sock_replicate_target , s_blob_receiver , ) . unwrap ( ) ; for _ in 0 .. 10 { let done = crdt_target . read ( ) . unwrap ( ) . update_index == 2 && crdt_leader . read ( ) . unwrap ( ) . update_index == 2 ; if done { break ; } let timer = Duration :: new ( 1 , 0 ) ; sleep ( timer ) ; } //send the data through let mut bq = VecDeque :: new ( ) ; let b = blob_recycler . allocate ( ) ; b . write ( ) . unwrap ( ) . meta . size = 10 ; bq . push_back ( b ) ; s_retransmit . send ( bq ) . unwrap ( ) ; let timer = Duration :: new ( 5 , 0 ) ; trace ! ( " " ) ; let mut oq = r_blob_receiver . recv_timeout ( timer ) . unwrap ( ) ; assert_eq ! ( oq len ( ) 1 ) ; let o = oq . pop_front ( ) . unwrap ( ) ; let ro = o . read ( ) . unwrap ( ) ; assert_eq ! ( ro meta size 10 ) ; assert_eq ! ( ro meta addr ( ) saddr ) ; exit . store ( true , Ordering :: Relaxed ) ; let threads = vec ! [ t_receiver t_retransmit t_crdt_target_g t_crdt_target_l t_crdt_leader_g t_crdt_leader_l ] ; for t in threads { t . join ( ) . unwrap ( ) ; } }
retransmit from leader to replicate target
retransmit from leader to replicate target