pub fn run(&mut self) {
        info!("running admin 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,
        ));

        // run in a loop, accepting new sessions and events on existing sessions
        loop {
            increment_counter!(&Stat::AdminEventLoop);

            if self.poll.poll(&mut events, timeout).is_err() {
                error!("Error polling");
            }

            increment_counter_by!(&Stat::AdminEventTotal, events.iter().count() as u64,);

            // handle all events
            for event in events.iter() {
                // handle new sessions
                if event.token() == Token(LISTENER_TOKEN) {
                    while let Ok((stream, addr)) = self.listener.accept() {
                        let mut session = if let Some(ssl_context) = &self.ssl_context {
                            // handle TLS if it is configured
                            match Ssl::new(&ssl_context).map(|v| v.accept(stream)) {
                                Ok(Ok(tls_stream)) => {
                                    // fully-negotiated session on accept()
                                    Session::tls(addr, tls_stream)
                                }
                                Ok(Err(HandshakeError::WouldBlock(tls_stream))) => {
                                    // session needs additional handshaking
                                    Session::handshaking(addr, tls_stream)
                                }
                                Ok(Err(_)) | Err(_) => {
                                    // unrecoverable error
                                    increment_counter!(&Stat::TcpAcceptEx);
                                    continue;
                                }
                            }
                        } else {
                            // plaintext session
                            Session::plain(addr, stream)
                        };

                        trace!("accepted new session: {}", addr);
                        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 {
                            increment_counter!(&Stat::TcpAcceptEx);
                        }
                    }
                } else {
                    // handle events on existing sessions
                    let token = event.token();
                    trace!("got event for admin session: {}", token.0);

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

                    // handle handshaking
                    if let Some(session) = self.sessions.get_mut(token.0) {
                        if session.is_handshaking() {
                            if let Err(e) = session.do_handshake() {
                                if e.kind() == ErrorKind::WouldBlock {
                                    // the session is still handshaking
                                    continue;
                                } else {
                                    // some error occured while handshaking
                                    self.close(token);
                                }
                            }
                        }
                    }

                    // handle write events before read events to reduce write
                    // buffer growth if there is also a readable event
                    if event.is_writable() {
                        increment_counter!(&Stat::AdminEventWrite);
                        self.do_write(token);
                    }

                    // read events are handled last
                    if event.is_readable() {
                        increment_counter!(&Stat::AdminEventRead);
                        let _ = self.do_read(token);
                    };
                }
            }

            // poll queue to receive new messages
            #[allow(clippy::never_loop)]
            while let Ok(message) = self.message_receiver.try_recv() {
                match message {
                    Message::Shutdown => {
                        return;
                    }
                }
            }

            self.get_rusage();
        }
    }
/// Runs the `Admin` in a loop, accepting new sessions for the admin
/// listener and handling events on existing sessions.
Runs the `Admin` in a loop, accepting new sessions for the admin
listener and handling events on existing sessions.
Runs the ` Admin ` in a loop accepting new sessions for the admin listener and handling events on existing sessions .
pub fn run(&mut self) {
        info!("running admin 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,
        ));
        // run in a loop, accepting new sessions and events on existing sessions
        loop {
            increment_counter!(&Stat::AdminEventLoop);
            if self.poll.poll(&mut events, timeout).is_err() {
                error!("Error polling");
            }
            increment_counter_by!(&Stat::AdminEventTotal, events.iter().count() as u64,);
            // handle all events
            for event in events.iter() {
                // handle new sessions
                if event.token() == Token(LISTENER_TOKEN) {
                    while let Ok((stream, addr)) = self.listener.accept() {
                        let mut session = if let Some(ssl_context) = &self.ssl_context {
                            // handle TLS if it is configured
                            match Ssl::new(&ssl_context).map(|v| v.accept(stream)) {
                                Ok(Ok(tls_stream)) => {
                                    // fully-negotiated session on accept()
                                    Session::tls(addr, tls_stream)
                                }
                                Ok(Err(HandshakeError::WouldBlock(tls_stream))) => {
                                    // session needs additional handshaking
                                    Session::handshaking(addr, tls_stream)
                                }
                                Ok(Err(_)) | Err(_) => {
                                    // unrecoverable error
                                    increment_counter!(&Stat::TcpAcceptEx);
                                    continue;
                                }
                            }
                        } else {
                            // plaintext session
                            Session::plain(addr, stream)
                        };
                        trace!("accepted new session: {}", addr);
                        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 {
                            increment_counter!(&Stat::TcpAcceptEx);
                        }
                    }
                } else {
                    // handle events on existing sessions
                    let token = event.token();
                    trace!("got event for admin session: {}", token.0);
                    // handle error events first
                    if event.is_error() {
                        increment_counter!(&Stat::AdminEventError);
                        self.handle_error(token);
                    }
                    // handle handshaking
                    if let Some(session) = self.sessions.get_mut(token.0) {
                        if session.is_handshaking() {
                            if let Err(e) = session.do_handshake() {
                                if e.kind() == ErrorKind::WouldBlock {
                                    // the session is still handshaking
                                    continue;
                                } else {
                                    // some error occured while handshaking
                                    self.close(token);
                                }
                            }
                        }
                    }
                    // handle write events before read events to reduce write
                    // buffer growth if there is also a readable event
                    if event.is_writable() {
                        increment_counter!(&Stat::AdminEventWrite);
                        self.do_write(token);
                    }
                    // read events are handled last
                    if event.is_readable() {
                        increment_counter!(&Stat::AdminEventRead);
                        let _ = self.do_read(token);
                    };
                }
            }
            // poll queue to receive new messages
            #[allow(clippy::never_loop)]
            while let Ok(message) = self.message_receiver.try_recv() {
                match message {
                    Message::Shutdown => {
                        return;
                    }
                }
            }
            self.get_rusage();
        }
    }
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 , ) ) ; // run in a loop, accepting new sessions and events on existing sessions loop { increment_counter ! ( Stat AdminEventLoop ) ; if self . poll . poll ( & mut events , timeout ) . is_err ( ) { error ! ( " " ) ; } increment_counter_by ! ( Stat AdminEventTotal events iter ( ) count ( ) as u64 ) ; // handle all events for event in events . iter ( ) { // handle new sessions if event . token ( ) == Token ( LISTENER_TOKEN ) { while let Ok ( ( stream , addr ) ) = self . listener . accept ( ) { let mut session = if let Some ( ssl_context ) = & self . ssl_context { // handle TLS if it is configured match Ssl :: new ( & ssl_context ) . map ( | v | v . accept ( stream ) ) { Ok ( Ok ( tls_stream ) ) => { // fully-negotiated session on accept() Session :: tls ( addr , tls_stream ) } Ok ( Err ( HandshakeError :: WouldBlock ( tls_stream ) ) ) => { // session needs additional handshaking Session :: handshaking ( addr , tls_stream ) } Ok ( Err ( _ ) ) | Err ( _ ) => { // unrecoverable error increment_counter ! ( Stat TcpAcceptEx ) ; continue ; } } } else { // plaintext session Session :: plain ( addr , stream ) } ; trace ! ( " " addr ) ; 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 { increment_counter ! ( Stat TcpAcceptEx ) ; } } } else { // handle events on existing sessions let token = event . token ( ) ; trace ! ( " " token 0 ) ; // handle error events first if event . is_error ( ) { increment_counter ! ( Stat AdminEventError ) ; self . handle_error ( token ) ; } // handle handshaking if let Some ( session ) = self . sessions . get_mut ( token . 0 ) { if session . is_handshaking ( ) { if let Err ( e ) = session . do_handshake ( ) { if e . kind ( ) == ErrorKind :: WouldBlock { // the session is still handshaking continue ; } else { // some error occured while handshaking self . close ( token ) ; } } } } // handle write events before read events to reduce write // buffer growth if there is also a readable event if event . is_writable ( ) { increment_counter ! ( Stat AdminEventWrite ) ; self . do_write ( token ) ; } // read events are handled last if event . is_readable ( ) { increment_counter ! ( Stat AdminEventRead ) ; let _ = self . do_read ( token ) ; } ; } } // poll queue to receive new messages # [ allow ( clippy never_loop ) ] while let Ok ( message ) = self . message_receiver . try_recv ( ) { match message { Message :: Shutdown => { return ; } } } self . get_rusage ( ) ; } }
Runs the `Admin` in a loop, accepting new sessions for the admin
listener and handling events on existing sessions.
Runs the ` Admin ` in a loop accepting new sessions for the admin listener and handling events on existing sessions .
&mut self
&mut self