pub fn serve<W: Write + Send + 'static>(
        obj: &SharedSkel,
        me: ReplicatedData,
        serve: UdpSocket,
        gossip: UdpSocket,
        exit: Arc<AtomicBool>,
        writer: W,
    ) -> Result<Vec<JoinHandle<()>>> {
        let crdt = Arc::new(RwLock::new(Crdt::new(me)));
        let t_gossip = Crdt::gossip(crdt.clone(), exit.clone());
        let t_listen = Crdt::listen(crdt.clone(), gossip, exit.clone());

        // make sure we are on the same interface
        let mut local = serve.local_addr()?;
        local.set_port(0);
        let respond_socket = UdpSocket::bind(local.clone())?;

        let packet_recycler = packet::PacketRecycler::default();
        let blob_recycler = packet::BlobRecycler::default();
        let (packet_sender, packet_receiver) = channel();
        let t_receiver =
            streamer::receiver(serve, exit.clone(), packet_recycler.clone(), packet_sender)?;
        let (responder_sender, responder_receiver) = channel();
        let t_responder = streamer::responder(
            respond_socket,
            exit.clone(),
            blob_recycler.clone(),
            responder_receiver,
        );
        let (verified_sender, verified_receiver) = channel();

        let mut verify_threads = Vec::new();
        let shared_verified_sender = Arc::new(Mutex::new(verified_sender));
        let shared_packet_receiver = Arc::new(Mutex::new(packet_receiver));
        for _ in 0..4 {
            let exit_ = exit.clone();
            let recv = shared_packet_receiver.clone();
            let sender = shared_verified_sender.clone();
            let thread = spawn(move || loop {
                let e = Self::verifier(&recv, &sender);
                if e.is_err() && exit_.load(Ordering::Relaxed) {
                    break;
                }
            });
            verify_threads.push(thread);
        }

        let (broadcast_sender, broadcast_receiver) = channel();

        let broadcast_socket = UdpSocket::bind(local)?;
        let t_broadcast = streamer::broadcaster(
            broadcast_socket,
            exit.clone(),
            crdt.clone(),
            blob_recycler.clone(),
            broadcast_receiver,
        );

        let t_sync = Self::sync_service(
            obj.clone(),
            exit.clone(),
            broadcast_sender,
            blob_recycler.clone(),
            Arc::new(Mutex::new(writer)),
        );

        let skel = obj.clone();
        let t_server = spawn(move || loop {
            let e = Self::process(
                &mut skel.clone(),
                &verified_receiver,
                &responder_sender,
                &packet_recycler,
                &blob_recycler,
            );
            if e.is_err() {
                if exit.load(Ordering::Relaxed) {
                    break;
                }
            }
        });

        let mut threads = vec![
            t_receiver,
            t_responder,
            t_server,
            t_sync,
            t_gossip,
            t_listen,
            t_broadcast,
        ];
        threads.extend(verify_threads.into_iter());
        Ok(threads)
    }
/// Create a UDP microservice that forwards messages the given AccountantSkel.
/// This service is the network leader
/// Set `exit` to shutdown its threads.
Create a UDP microservice that forwards messages the given AccountantSkel.
This service is the network leader
Set `exit` to shutdown its threads.
Create a UDP microservice that forwards messages the given AccountantSkel . This service is the network leader Set ` exit ` to shutdown its threads .
pub fn serve<W: Write + Send + 'static>(
        obj: &SharedSkel,
        me: ReplicatedData,
        serve: UdpSocket,
        gossip: UdpSocket,
        exit: Arc<AtomicBool>,
        writer: W,
    ) -> Result<Vec<JoinHandle<()>>> {
        let crdt = Arc::new(RwLock::new(Crdt::new(me)));
        let t_gossip = Crdt::gossip(crdt.clone(), exit.clone());
        let t_listen = Crdt::listen(crdt.clone(), gossip, exit.clone());
        // make sure we are on the same interface
        let mut local = serve.local_addr()?;
        local.set_port(0);
        let respond_socket = UdpSocket::bind(local.clone())?;
        let packet_recycler = packet::PacketRecycler::default();
        let blob_recycler = packet::BlobRecycler::default();
        let (packet_sender, packet_receiver) = channel();
        let t_receiver =
            streamer::receiver(serve, exit.clone(), packet_recycler.clone(), packet_sender)?;
        let (responder_sender, responder_receiver) = channel();
        let t_responder = streamer::responder(
            respond_socket,
            exit.clone(),
            blob_recycler.clone(),
            responder_receiver,
        );
        let (verified_sender, verified_receiver) = channel();
        let mut verify_threads = Vec::new();
        let shared_verified_sender = Arc::new(Mutex::new(verified_sender));
        let shared_packet_receiver = Arc::new(Mutex::new(packet_receiver));
        for _ in 0..4 {
            let exit_ = exit.clone();
            let recv = shared_packet_receiver.clone();
            let sender = shared_verified_sender.clone();
            let thread = spawn(move || loop {
                let e = Self::verifier(&recv, &sender);
                if e.is_err() && exit_.load(Ordering::Relaxed) {
                    break;
                }
            });
            verify_threads.push(thread);
        }
        let (broadcast_sender, broadcast_receiver) = channel();
        let broadcast_socket = UdpSocket::bind(local)?;
        let t_broadcast = streamer::broadcaster(
            broadcast_socket,
            exit.clone(),
            crdt.clone(),
            blob_recycler.clone(),
            broadcast_receiver,
        );
        let t_sync = Self::sync_service(
            obj.clone(),
            exit.clone(),
            broadcast_sender,
            blob_recycler.clone(),
            Arc::new(Mutex::new(writer)),
        );
        let skel = obj.clone();
        let t_server = spawn(move || loop {
            let e = Self::process(
                &mut skel.clone(),
                &verified_receiver,
                &responder_sender,
                &packet_recycler,
                &blob_recycler,
            );
            if e.is_err() {
                if exit.load(Ordering::Relaxed) {
                    break;
                }
            }
        });
        let mut threads = vec![
            t_receiver,
            t_responder,
            t_server,
            t_sync,
            t_gossip,
            t_listen,
            t_broadcast,
        ];
        threads.extend(verify_threads.into_iter());
        Ok(threads)
    }
pub fn serve < W : Write + Send + ' static > ( obj : & SharedSkel , me : ReplicatedData , serve : UdpSocket , gossip : UdpSocket , exit : Arc < AtomicBool > , writer : W , ) -> Result < Vec < JoinHandle < ( ) > > > { let crdt = Arc :: new ( RwLock :: new ( Crdt :: new ( me ) ) ) ; let t_gossip = Crdt :: gossip ( crdt . clone ( ) , exit . clone ( ) ) ; let t_listen = Crdt :: listen ( crdt . clone ( ) , gossip , exit . clone ( ) ) ; // make sure we are on the same interface let mut local = serve . local_addr ( ) ? ; local . set_port ( 0 ) ; let respond_socket = UdpSocket :: bind ( local . clone ( ) ) ? ; let packet_recycler = packet :: PacketRecycler :: default ( ) ; let blob_recycler = packet :: BlobRecycler :: default ( ) ; let ( packet_sender , packet_receiver ) = channel ( ) ; let t_receiver = streamer :: receiver ( serve , exit . clone ( ) , packet_recycler . clone ( ) , packet_sender ) ? ; let ( responder_sender , responder_receiver ) = channel ( ) ; let t_responder = streamer :: responder ( respond_socket , exit . clone ( ) , blob_recycler . clone ( ) , responder_receiver , ) ; let ( verified_sender , verified_receiver ) = channel ( ) ; let mut verify_threads = Vec :: new ( ) ; let shared_verified_sender = Arc :: new ( Mutex :: new ( verified_sender ) ) ; let shared_packet_receiver = Arc :: new ( Mutex :: new ( packet_receiver ) ) ; for _ in 0 .. 4 { let exit_ = exit . clone ( ) ; let recv = shared_packet_receiver . clone ( ) ; let sender = shared_verified_sender . clone ( ) ; let thread = spawn ( move | | loop { let e = Self :: verifier ( & recv , & sender ) ; if e . is_err ( ) && exit_ . load ( Ordering :: Relaxed ) { break ; } } ) ; verify_threads . push ( thread ) ; } let ( broadcast_sender , broadcast_receiver ) = channel ( ) ; let broadcast_socket = UdpSocket :: bind ( local ) ? ; let t_broadcast = streamer :: broadcaster ( broadcast_socket , exit . clone ( ) , crdt . clone ( ) , blob_recycler . clone ( ) , broadcast_receiver , ) ; let t_sync = Self :: sync_service ( obj . clone ( ) , exit . clone ( ) , broadcast_sender , blob_recycler . clone ( ) , Arc :: new ( Mutex :: new ( writer ) ) , ) ; let skel = obj . clone ( ) ; let t_server = spawn ( move | | loop { let e = Self :: process ( & mut skel . clone ( ) , & verified_receiver , & responder_sender , & packet_recycler , & blob_recycler , ) ; if e . is_err ( ) { if exit . load ( Ordering :: Relaxed ) { break ; } } } ) ; let mut threads = vec ! [ t_receiver t_responder t_server t_sync t_gossip t_listen t_broadcast ] ; threads . extend ( verify_threads . into_iter ( ) ) ; Ok ( threads ) }
Create a UDP microservice that forwards messages the given AccountantSkel.
Create a UDP microservice that forwards messages the given AccountantSkel .
obj
&SharedSkel
me
ReplicatedData
serve
UdpSocket
gossip
UdpSocket
exit
Arc<AtomicBool>
writer
W
obj
&SharedSkel
me
ReplicatedData
serve
UdpSocket
gossip
UdpSocket
exit
Arc<AtomicBool>
writer
W