async fn drive_peer_request(&mut self, req: Request) {
        trace!(?req);
        use tower::{load_shed::error::Overloaded, ServiceExt};

        // Add a metric for inbound requests
        metrics::counter!(
            "zebra.net.in.requests",
            1,
            "command" => req.command(),
            "addr" => self.metrics_label.clone(),
        );
        self.update_state_metrics(format!("In::Req::{}", req.command()));

        if self.svc.ready().await.is_err() {
            // Treat all service readiness errors as Overloaded
            // TODO: treat `TryRecvError::Closed` in `Inbound::poll_ready` as a fatal error (#1655)
            self.fail_with(PeerError::Overloaded);
            return;
        }

        let rsp = match self.svc.call(req.clone()).await {
            Err(e) => {
                if e.is::<Overloaded>() {
                    tracing::info!("inbound service is overloaded, closing connection");
                    metrics::counter!("pool.closed.loadshed", 1);
                    self.fail_with(PeerError::Overloaded);
                } else {
                    // We could send a reject to the remote peer, but that might cause
                    // them to disconnect, and we might be using them to sync blocks.
                    // For similar reasons, we don't want to fail_with() here - we
                    // only close the connection if the peer is doing something wrong.
                    info!(%e,
                          connection_state = ?self.state,
                          client_receiver = ?self.client_rx,
                          "error processing peer request");
                }
                return;
            }
            Ok(rsp) => rsp,
        };

        // Add a metric for outbound responses to inbound requests
        metrics::counter!(
            "zebra.net.out.responses",
            1,
            "command" => rsp.command(),
            "addr" => self.metrics_label.clone(),
        );
        self.update_state_metrics(format!("In::Rsp::{}", rsp.command()));

        match rsp.clone() {
            Response::Nil => { /* generic success, do nothing */ }
            Response::Peers(addrs) => {
                if let Err(e) = self.peer_tx.send(Message::Addr(addrs)).await {
                    self.fail_with(e);
                }
            }
            Response::Transactions(transactions) => {
                // Generate one tx message per transaction.
                for transaction in transactions.into_iter() {
                    if let Err(e) = self.peer_tx.send(Message::Tx(transaction)).await {
                        self.fail_with(e);
                        return;
                    }
                }
            }
            Response::Blocks(blocks) => {
                // Generate one block message per block.
                for block in blocks.into_iter() {
                    if let Err(e) = self.peer_tx.send(Message::Block(block)).await {
                        self.fail_with(e);
                        return;
                    }
                }
            }
            Response::BlockHashes(hashes) => {
                if let Err(e) = self
                    .peer_tx
                    .send(Message::Inv(hashes.into_iter().map(Into::into).collect()))
                    .await
                {
                    self.fail_with(e)
                }
            }
            Response::BlockHeaders(headers) => {
                if let Err(e) = self.peer_tx.send(Message::Headers(headers)).await {
                    self.fail_with(e)
                }
            }
            Response::TransactionIds(hashes) => {
                if let Err(e) = self
                    .peer_tx
                    .send(Message::Inv(hashes.into_iter().map(Into::into).collect()))
                    .await
                {
                    self.fail_with(e)
                }
            }
        }

        debug!(state = %self.state, %req, %rsp, "sent Zebra response to peer");
    }
/// Given a `req` originating from the peer, drive it to completion and send
/// any appropriate messages to the remote peer. If an error occurs while
/// processing the request (e.g., the service is shedding load), then we call
/// fail_with to terminate the entire peer connection, shrinking the number
/// of connected peers.
Given a `req` originating from the peer, drive it to completion and send
any appropriate messages to the remote peer. If an error occurs while
processing the request , then we call
fail_with to terminate the entire peer connection, shrinking the number
of connected peers.
Given a ` req ` originating from the peer drive it to completion and send any appropriate messages to the remote peer . If an error occurs while processing the request then we call fail_with to terminate the entire peer connection shrinking the number of connected peers .
async fn drive_peer_request(&mut self, req: Request) {
        trace!(?req);
        use tower::{load_shed::error::Overloaded, ServiceExt};
        // Add a metric for inbound requests
        metrics::counter!(
            "zebra.net.in.requests",
            1,
            "command" => req.command(),
            "addr" => self.metrics_label.clone(),
        );
        self.update_state_metrics(format!("In::Req::{}", req.command()));
        if self.svc.ready().await.is_err() {
            // Treat all service readiness errors as Overloaded
            // TODO: treat `TryRecvError::Closed` in `Inbound::poll_ready` as a fatal error (#1655)
            self.fail_with(PeerError::Overloaded);
            return;
        }
        let rsp = match self.svc.call(req.clone()).await {
            Err(e) => {
                if e.is::<Overloaded>() {
                    tracing::info!("inbound service is overloaded, closing connection");
                    metrics::counter!("pool.closed.loadshed", 1);
                    self.fail_with(PeerError::Overloaded);
                } else {
                    // We could send a reject to the remote peer, but that might cause
                    // them to disconnect, and we might be using them to sync blocks.
                    // For similar reasons, we don't want to fail_with() here - we
                    // only close the connection if the peer is doing something wrong.
                    info!(%e,
                          connection_state = ?self.state,
                          client_receiver = ?self.client_rx,
                          "error processing peer request");
                }
                return;
            }
            Ok(rsp) => rsp,
        };
        // Add a metric for outbound responses to inbound requests
        metrics::counter!(
            "zebra.net.out.responses",
            1,
            "command" => rsp.command(),
            "addr" => self.metrics_label.clone(),
        );
        self.update_state_metrics(format!("In::Rsp::{}", rsp.command()));
        match rsp.clone() {
            Response::Nil => { /* generic success, do nothing */ }
            Response::Peers(addrs) => {
                if let Err(e) = self.peer_tx.send(Message::Addr(addrs)).await {
                    self.fail_with(e);
                }
            }
            Response::Transactions(transactions) => {
                // Generate one tx message per transaction.
                for transaction in transactions.into_iter() {
                    if let Err(e) = self.peer_tx.send(Message::Tx(transaction)).await {
                        self.fail_with(e);
                        return;
                    }
                }
            }
            Response::Blocks(blocks) => {
                // Generate one block message per block.
                for block in blocks.into_iter() {
                    if let Err(e) = self.peer_tx.send(Message::Block(block)).await {
                        self.fail_with(e);
                        return;
                    }
                }
            }
            Response::BlockHashes(hashes) => {
                if let Err(e) = self
                    .peer_tx
                    .send(Message::Inv(hashes.into_iter().map(Into::into).collect()))
                    .await
                {
                    self.fail_with(e)
                }
            }
            Response::BlockHeaders(headers) => {
                if let Err(e) = self.peer_tx.send(Message::Headers(headers)).await {
                    self.fail_with(e)
                }
            }
            Response::TransactionIds(hashes) => {
                if let Err(e) = self
                    .peer_tx
                    .send(Message::Inv(hashes.into_iter().map(Into::into).collect()))
                    .await
                {
                    self.fail_with(e)
                }
            }
        }
        debug!(state = %self.state, %req, %rsp, "sent Zebra response to peer");
    }
async fn drive_peer_request ( & mut self , req : Request ) { trace ! ( req ) ; use tower :: { load_shed :: error :: Overloaded , ServiceExt } ; // Add a metric for inbound requests metrics :: counter ! ( " " 1 " " req command ( ) " " self metrics_label clone ( ) ) ; self . update_state_metrics ( format ! ( " " req command ( ) ) ) ; if self . svc . ready ( ) . await . is_err ( ) { // Treat all service readiness errors as Overloaded // TODO: treat `TryRecvError::Closed` in `Inbound::poll_ready` as a fatal error (#1655) self . fail_with ( PeerError :: Overloaded ) ; return ; } let rsp = match self . svc . call ( req . clone ( ) ) . await { Err ( e ) => { if e . is :: < Overloaded > ( ) { tracing :: info ! ( " " ) ; metrics :: counter ! ( " " 1 ) ; self . fail_with ( PeerError :: Overloaded ) ; } else { // We could send a reject to the remote peer, but that might cause // them to disconnect, and we might be using them to sync blocks. // For similar reasons, we don't want to fail_with() here - we // only close the connection if the peer is doing something wrong. info ! ( e connection_state self state client_receiver self client_rx " " ) ; } return ; } Ok ( rsp ) => rsp , } ; // Add a metric for outbound responses to inbound requests metrics :: counter ! ( " " 1 " " rsp command ( ) " " self metrics_label clone ( ) ) ; self . update_state_metrics ( format ! ( " " rsp command ( ) ) ) ; match rsp . clone ( ) { Response :: Nil => { /* generic success, do nothing */ } Response :: Peers ( addrs ) => { if let Err ( e ) = self . peer_tx . send ( Message :: Addr ( addrs ) ) . await { self . fail_with ( e ) ; } } Response :: Transactions ( transactions ) => { // Generate one tx message per transaction. for transaction in transactions . into_iter ( ) { if let Err ( e ) = self . peer_tx . send ( Message :: Tx ( transaction ) ) . await { self . fail_with ( e ) ; return ; } } } Response :: Blocks ( blocks ) => { // Generate one block message per block. for block in blocks . into_iter ( ) { if let Err ( e ) = self . peer_tx . send ( Message :: Block ( block ) ) . await { self . fail_with ( e ) ; return ; } } } Response :: BlockHashes ( hashes ) => { if let Err ( e ) = self . peer_tx . send ( Message :: Inv ( hashes . into_iter ( ) . map ( Into :: into ) . collect ( ) ) ) . await { self . fail_with ( e ) } } Response :: BlockHeaders ( headers ) => { if let Err ( e ) = self . peer_tx . send ( Message :: Headers ( headers ) ) . await { self . fail_with ( e ) } } Response :: TransactionIds ( hashes ) => { if let Err ( e ) = self . peer_tx . send ( Message :: Inv ( hashes . into_iter ( ) . map ( Into :: into ) . collect ( ) ) ) . await { self . fail_with ( e ) } } } debug ! ( state self state req rsp " " ) ; }
Given a `req` originating from the peer, drive it to completion and send
any appropriate messages to the remote peer.
Given a ` req ` originating from the peer drive it to completion and send any appropriate messages to the remote peer .
&mut self
req
Request