pub fn run(&mut self) {
        info!("running server on: {}", self.addr);

        let mut events = Events::with_capacity(self.config.server().nevent());
        let timeout = Some(std::time::Duration::from_millis(
            self.config.server().timeout() as u64,
        ));

        // repeatedly run accepting new connections and moving them to the worker
        loop {
            increment_counter!(&Stat::ServerEventLoop);
            if self.poll.poll(&mut events, timeout).is_err() {
                error!("Error polling server");
            }
            increment_counter_by!(&Stat::ServerEventTotal, events.iter().count() as u64,);

            // handle all events
            for event in events.iter() {
                if event.token() == Token(LISTENER_TOKEN) {
                    while let Ok((stream, addr)) = self.listener.accept() {
                        // handle TLS if it is configured
                        if let Some(ssl_context) = &self.ssl_context {
                            match Ssl::new(&ssl_context).map(|v| v.accept(stream)) {
                                // handle case where we have a fully-negotiated
                                // TLS stream on accept()
                                Ok(Ok(tls_stream)) => {
                                    let session = Session::tls(addr, tls_stream);
                                    trace!("accepted new session: {}", addr);
                                    if self.sender.send(session).is_err() {
                                        error!("error sending session to worker");
                                        let _ = increment_counter!(&Stat::TcpAcceptEx);
                                    }
                                }
                                // handle case where further negotiation is
                                // needed
                                Ok(Err(HandshakeError::WouldBlock(tls_stream))) => {
                                    let mut session = Session::handshaking(addr, tls_stream);
                                    let s = self.sessions.vacant_entry();
                                    let token = s.key();
                                    session.set_token(Token(token));
                                    if session.register(&self.poll).is_ok() {
                                        s.insert(session);
                                    } else {
                                        let _ = increment_counter!(&Stat::TcpAcceptEx);
                                    }
                                }
                                // some other error has occurred and we drop the
                                // stream
                                Ok(Err(_)) | Err(_) => {
                                    increment_counter!(&Stat::TcpAcceptEx);
                                }
                            }
                        } else {
                            let session = Session::plain(addr, stream);
                            trace!("accepted new session: {}", addr);
                            if self.sender.send(session).is_err() {
                                error!("error sending session to worker");
                                increment_counter!(&Stat::TcpAcceptEx);
                            }
                        };
                    }
                } else {
                    // handle plain sessions
                    let token = event.token();
                    trace!("got event for session: {}", token.0);

                    // handle error events first
                    if event.is_error() {
                        increment_counter!(&Stat::ServerEventError);
                        self.handle_error(token);
                    }

                    if let Some(session) = self.sessions.get_mut(token.0) {
                        if session.do_handshake().is_ok() {
                            let mut session = self.sessions.remove(token.0);
                            let _ = session.deregister(&self.poll);
                            if self.sender.send(session).is_err() {
                                error!("error sending session to worker");
                                increment_counter!(&Stat::TcpAcceptEx);
                            }
                        }
                    }
                }
            }

            // poll queue to receive new messages
            #[allow(clippy::never_loop)]
            while let Ok(message) = self.message_receiver.try_recv() {
                match message {
                    Message::Shutdown => {
                        return;
                    }
                }
            }
        }
    }
/// Runs the `Server` in a loop, accepting new sessions and moving them to
/// the queue
Runs the `Server` in a loop, accepting new sessions and moving them to
the queue
Runs the ` Server ` in a loop accepting new sessions and moving them to the queue
pub fn run(&mut self) {
        info!("running server on: {}", self.addr);
        let mut events = Events::with_capacity(self.config.server().nevent());
        let timeout = Some(std::time::Duration::from_millis(
            self.config.server().timeout() as u64,
        ));
        // repeatedly run accepting new connections and moving them to the worker
        loop {
            increment_counter!(&Stat::ServerEventLoop);
            if self.poll.poll(&mut events, timeout).is_err() {
                error!("Error polling server");
            }
            increment_counter_by!(&Stat::ServerEventTotal, events.iter().count() as u64,);
            // handle all events
            for event in events.iter() {
                if event.token() == Token(LISTENER_TOKEN) {
                    while let Ok((stream, addr)) = self.listener.accept() {
                        // handle TLS if it is configured
                        if let Some(ssl_context) = &self.ssl_context {
                            match Ssl::new(&ssl_context).map(|v| v.accept(stream)) {
                                // handle case where we have a fully-negotiated
                                // TLS stream on accept()
                                Ok(Ok(tls_stream)) => {
                                    let session = Session::tls(addr, tls_stream);
                                    trace!("accepted new session: {}", addr);
                                    if self.sender.send(session).is_err() {
                                        error!("error sending session to worker");
                                        let _ = increment_counter!(&Stat::TcpAcceptEx);
                                    }
                                }
                                // handle case where further negotiation is
                                // needed
                                Ok(Err(HandshakeError::WouldBlock(tls_stream))) => {
                                    let mut session = Session::handshaking(addr, tls_stream);
                                    let s = self.sessions.vacant_entry();
                                    let token = s.key();
                                    session.set_token(Token(token));
                                    if session.register(&self.poll).is_ok() {
                                        s.insert(session);
                                    } else {
                                        let _ = increment_counter!(&Stat::TcpAcceptEx);
                                    }
                                }
                                // some other error has occurred and we drop the
                                // stream
                                Ok(Err(_)) | Err(_) => {
                                    increment_counter!(&Stat::TcpAcceptEx);
                                }
                            }
                        } else {
                            let session = Session::plain(addr, stream);
                            trace!("accepted new session: {}", addr);
                            if self.sender.send(session).is_err() {
                                error!("error sending session to worker");
                                increment_counter!(&Stat::TcpAcceptEx);
                            }
                        };
                    }
                } else {
                    // handle plain sessions
                    let token = event.token();
                    trace!("got event for session: {}", token.0);
                    // handle error events first
                    if event.is_error() {
                        increment_counter!(&Stat::ServerEventError);
                        self.handle_error(token);
                    }
                    if let Some(session) = self.sessions.get_mut(token.0) {
                        if session.do_handshake().is_ok() {
                            let mut session = self.sessions.remove(token.0);
                            let _ = session.deregister(&self.poll);
                            if self.sender.send(session).is_err() {
                                error!("error sending session to worker");
                                increment_counter!(&Stat::TcpAcceptEx);
                            }
                        }
                    }
                }
            }
            // poll queue to receive new messages
            #[allow(clippy::never_loop)]
            while let Ok(message) = self.message_receiver.try_recv() {
                match message {
                    Message::Shutdown => {
                        return;
                    }
                }
            }
        }
    }
pub fn run ( & mut self ) { info ! ( " " self addr ) ; let mut events = Events :: with_capacity ( self . config . server ( ) . nevent ( ) ) ; let timeout = Some ( std :: time :: Duration :: from_millis ( self . config . server ( ) . timeout ( ) as u64 , ) ) ; // repeatedly run accepting new connections and moving them to the worker loop { increment_counter ! ( Stat ServerEventLoop ) ; if self . poll . poll ( & mut events , timeout ) . is_err ( ) { error ! ( " " ) ; } increment_counter_by ! ( Stat ServerEventTotal events iter ( ) count ( ) as u64 ) ; // handle all events for event in events . iter ( ) { if event . token ( ) == Token ( LISTENER_TOKEN ) { while let Ok ( ( stream , addr ) ) = self . listener . accept ( ) { // handle TLS if it is configured if let Some ( ssl_context ) = & self . ssl_context { match Ssl :: new ( & ssl_context ) . map ( | v | v . accept ( stream ) ) { // handle case where we have a fully-negotiated // TLS stream on accept() Ok ( Ok ( tls_stream ) ) => { let session = Session :: tls ( addr , tls_stream ) ; trace ! ( " " addr ) ; if self . sender . send ( session ) . is_err ( ) { error ! ( " " ) ; let _ = increment_counter ! ( Stat TcpAcceptEx ) ; } } // handle case where further negotiation is // needed Ok ( Err ( HandshakeError :: WouldBlock ( tls_stream ) ) ) => { let mut session = Session :: handshaking ( addr , tls_stream ) ; let s = self . sessions . vacant_entry ( ) ; let token = s . key ( ) ; session . set_token ( Token ( token ) ) ; if session . register ( & self . poll ) . is_ok ( ) { s . insert ( session ) ; } else { let _ = increment_counter ! ( Stat TcpAcceptEx ) ; } } // some other error has occurred and we drop the // stream Ok ( Err ( _ ) ) | Err ( _ ) => { increment_counter ! ( Stat TcpAcceptEx ) ; } } } else { let session = Session :: plain ( addr , stream ) ; trace ! ( " " addr ) ; if self . sender . send ( session ) . is_err ( ) { error ! ( " " ) ; increment_counter ! ( Stat TcpAcceptEx ) ; } } ; } } else { // handle plain sessions let token = event . token ( ) ; trace ! ( " " token 0 ) ; // handle error events first if event . is_error ( ) { increment_counter ! ( Stat ServerEventError ) ; self . handle_error ( token ) ; } if let Some ( session ) = self . sessions . get_mut ( token . 0 ) { if session . do_handshake ( ) . is_ok ( ) { let mut session = self . sessions . remove ( token . 0 ) ; let _ = session . deregister ( & self . poll ) ; if self . sender . send ( session ) . is_err ( ) { error ! ( " " ) ; increment_counter ! ( Stat TcpAcceptEx ) ; } } } } } // poll queue to receive new messages # [ allow ( clippy never_loop ) ] while let Ok ( message ) = self . message_receiver . try_recv ( ) { match message { Message :: Shutdown => { return ; } } } } }
Runs the `Server` in a loop, accepting new sessions and moving them to
the queue
Runs the ` Server ` in a loop accepting new sessions and moving them to the queue
&mut self
&mut self