/// Start to receive packets from some endpoint.
/// This function takes in a closure with which a packet contents will be asserted.
Start to receive packets from some endpoint.
This function takes in a closure with which a packet contents will be asserted.
Start to receive packets from some endpoint . This function takes in a closure with which a packet contents will be asserted .
pub fn start_receiving<F>(self, packet_assert: F) -> ServerHandle
    where
        F: Fn(Packet) + Send + Sized + 'static,
    {
        let mut socket = Socket::bind(self.listening_host).unwrap();
        let (notify_tx, notify_rx) = crossbeam_channel::unbounded();
        let (events_tx, events_rx) = crossbeam_channel::unbounded();
        let mut throughput_monitor = self.throughput_monitor;
        let serve_handle = thread::spawn(move || {
            loop {
                socket.manual_poll(Instant::now());
                match socket.recv() {
                    Some(result) => match result {
                        SocketEvent::Packet(p) => {
                            packet_assert(p);
                            if throughput_monitor.tick() {
                                if let Err(e) = events_tx.send(ServerEvent::Throughput(
                                    throughput_monitor.last_throughput(),
                                )) {
                                    error!("Client can not send packet {:?}", e);
                                }
                            }
                        }
                        _ => {
                            if let Err(e) = events_tx.send(ServerEvent::SocketEvent(result)) {
                                error!("Client cannot send packet {:?}", e);
                            }
                        }
                    },
                    None => {
                        // check if we received a notify to close the server.
                        match notify_rx.try_recv() {
                            Ok(notify) => match notify {
                                ServerCommand::Shutdown => {
                                    let result = || -> Result<(), crossbeam_channel::SendError<ServerEvent>> {
                                        events_tx.send(ServerEvent::AverageThroughput(
                                            throughput_monitor.average(),
                                        ))?;
                                        events_tx.send(ServerEvent::TotalSent(
                                            throughput_monitor.total_measured_ticks(),
                                        ))?;
                                        Ok(())
                                    };
                                    if let Err(e) = result() {
                                        error!("Unable to sent an event {:?}", e);
                                    };
                                    return;
                                }
                            },
                            Err(e) => {
                                if !e.is_empty() {
                                    error!(
                                        "Error occurred when trying to receive on notify channel"
                                    );
                                }
                            }
                        }
                    }
                }
            }
        });
        ServerHandle::new(serve_handle, notify_tx, events_rx, self.listening_host)
    }
pub fn start_receiving < F > ( self , packet_assert : F ) -> ServerHandle where F : Fn ( Packet ) + Send + Sized + ' static , { let mut socket = Socket :: bind ( self . listening_host ) . unwrap ( ) ; let ( notify_tx , notify_rx ) = crossbeam_channel :: unbounded ( ) ; let ( events_tx , events_rx ) = crossbeam_channel :: unbounded ( ) ; let mut throughput_monitor = self . throughput_monitor ; let serve_handle = thread :: spawn ( move | | { loop { socket . manual_poll ( Instant :: now ( ) ) ; match socket . recv ( ) { Some ( result ) => match result { SocketEvent :: Packet ( p ) => { packet_assert ( p ) ; if throughput_monitor . tick ( ) { if let Err ( e ) = events_tx . send ( ServerEvent :: Throughput ( throughput_monitor . last_throughput ( ) , ) ) { error ! ( " " e ) ; } } } _ => { if let Err ( e ) = events_tx . send ( ServerEvent :: SocketEvent ( result ) ) { error ! ( " " e ) ; } } } , None => { // check if we received a notify to close the server. match notify_rx . try_recv ( ) { Ok ( notify ) => match notify { ServerCommand :: Shutdown => { let result = | | -> Result < ( ) , crossbeam_channel :: SendError < ServerEvent > > { events_tx . send ( ServerEvent :: AverageThroughput ( throughput_monitor . average ( ) , ) ) ? ; events_tx . send ( ServerEvent :: TotalSent ( throughput_monitor . total_measured_ticks ( ) , ) ) ? ; Ok ( ( ) ) } ; if let Err ( e ) = result ( ) { error ! ( " " e ) ; } ; return ; } } , Err ( e ) => { if ! e . is_empty ( ) { error ! ( " " ) ; } } } } } } } ) ; ServerHandle :: new ( serve_handle , notify_tx , events_rx , self . listening_host ) }
Start to receive packets from some endpoint.
Start to receive packets from some endpoint .
self
packet_assert
F