pub async fn run<Rx>(mut self, mut peer_rx: Rx)
    where
        Rx: Stream<Item = Result<Message, SerializationError>> + Unpin,
    {
        // At a high level, the event loop we want is as follows: we check for any
        // incoming messages from the remote peer, check if they should be interpreted
        // as a response to a pending client request, and if not, interpret them as a
        // request from the remote peer to our node.
        //
        // We also need to handle those client requests in the first place. The client
        // requests are received from the corresponding `peer::Client` over a bounded
        // channel (with bound 1, to minimize buffering), but there is no relationship
        // between the stream of client requests and the stream of peer messages, so we
        // cannot ignore one kind while waiting on the other. Moreover, we cannot accept
        // a second client request while the first one is still pending.
        //
        // To do this, we inspect the current request state.
        //
        // If there is no pending request, we wait on either an incoming peer message or
        // an incoming request, whichever comes first.
        //
        // If there is a pending request, we wait only on an incoming peer message, and
        // check whether it can be interpreted as a response to the pending request.
        loop {
            match self.state {
                State::AwaitingRequest => {
                    trace!("awaiting client request or peer message");
                    // CORRECTNESS
                    //
                    // Currently, select prefers the first future if multiple
                    // futures are ready.
                    //
                    // The peer can starve client requests if it sends an
                    // uninterrupted series of messages. But this is unlikely in
                    // practice, due to network delays.
                    //
                    // If both futures are ready, there's no particular reason
                    // to prefer one over the other.
                    //
                    // TODO: use `futures::select!`, which chooses a ready future
                    //       at random, avoiding starvation
                    //       (To use `select!`, we'll need to map the different
                    //       results to a new enum types.)
                    match future::select(peer_rx.next(), self.client_rx.next()).await {
                        Either::Left((None, _)) => {
                            self.fail_with(PeerError::ConnectionClosed);
                        }
                        Either::Left((Some(Err(e)), _)) => self.fail_with(e),
                        Either::Left((Some(Ok(msg)), _)) => {
                            self.handle_message_as_request(msg).await
                        }
                        Either::Right((None, _)) => {
                            trace!("client_rx closed, ending connection");
                            return;
                        }
                        Either::Right((Some(req), _)) => {
                            let span = req.span.clone();
                            self.handle_client_request(req).instrument(span).await
                        }
                    }
                }
                // We're awaiting a response to a client request,
                // so wait on either a peer message, or on a request cancellation.
                State::AwaitingResponse {
                    ref span,
                    ref mut tx,
                    ..
                } => {
                    // we have to get rid of the span reference so we can tamper with the state
                    let span = span.clone();
                    trace!(parent: &span, "awaiting response to client request");
                    let timer_ref = self
                        .request_timer
                        .as_mut()
                        .expect("timeout must be set while awaiting response");
                    // CORRECTNESS
                    //
                    // Currently, select prefers the first future if multiple
                    // futures are ready.
                    //
                    // If multiple futures are ready, we want the cancellation
                    // to take priority, then the timeout, then peer responses.
                    let cancel = future::select(tx.cancellation(), timer_ref);
                    match future::select(cancel, peer_rx.next())
                        .instrument(span.clone())
                        .await
                    {
                        Either::Right((None, _)) => self.fail_with(PeerError::ConnectionClosed),
                        Either::Right((Some(Err(e)), _)) => self.fail_with(e),
                        Either::Right((Some(Ok(peer_msg)), _cancel)) => {
                            // Try to process the message using the handler.
                            // This extremely awkward construction avoids
                            // keeping a live reference to handler across the
                            // call to handle_message_as_request, which takes
                            // &mut self. This is a sign that we don't properly
                            // factor the state required for inbound and
                            // outbound requests.
                            let request_msg = match self.state {
                                State::AwaitingResponse {
                                    ref mut handler, ..
                                } => span.in_scope(|| handler.process_message(peer_msg)),
                                _ => unreachable!("unexpected state after AwaitingResponse: {:?}, peer_msg: {:?}, client_receiver: {:?}",
                                                  self.state,
                                                  peer_msg,
                                                  self.client_rx,
                                ),
                            };
                            // If the message was not consumed, check whether it
                            // should be handled as a request.
                            if let Some(msg) = request_msg {
                                // do NOT instrument with the request span, this is
                                // independent work
                                self.handle_message_as_request(msg).await;
                            } else {
                                // Otherwise, check whether the handler is finished
                                // processing messages and update the state.
                                self.state = match self.state {
                                    State::AwaitingResponse {
                                        handler: Handler::Finished(response),
                                        tx,
                                        ..
                                    } => {
                                        let _ = tx.send(response.map_err(Into::into));
                                        State::AwaitingRequest
                                    }
                                    pending @ State::AwaitingResponse { .. } => pending,
                                    _ => unreachable!(
                                        "unexpected failed connection state while AwaitingResponse: client_receiver: {:?}",
                                        self.client_rx
                                    ),
                                };
                            }
                        }
                        Either::Left((Either::Right(_), _peer_fut)) => {
                            trace!(parent: &span, "client request timed out");
                            let e = PeerError::ClientRequestTimeout;
                            self.state = match self.state {
                                // Special case: ping timeouts fail the connection.
                                State::AwaitingResponse {
                                    handler: Handler::Ping(_),
                                    ..
                                } => {
                                    self.fail_with(e);
                                    State::Failed
                                }
                                // Other request timeouts fail the request.
                                State::AwaitingResponse { tx, .. } => {
                                    let _ = tx.send(Err(e.into()));
                                    State::AwaitingRequest
                                }
                                _ => unreachable!(
                                    "unexpected failed connection state while AwaitingResponse: client_receiver: {:?}",
                                    self.client_rx
                                ),
                            };
                        }
                        Either::Left((Either::Left(_), _peer_fut)) => {
                            trace!(parent: &span, "client request was cancelled");
                            self.state = State::AwaitingRequest;
                        }
                    }
                }
                // We've failed, but we need to flush all pending client
                // requests before we can return and complete the future.
                State::Failed => {
                    match self.client_rx.next().await {
                        Some(InProgressClientRequest { tx, span, .. }) => {
                            trace!(
                                parent: &span,
                                "sending an error response to a pending request on a failed connection"
                            );
                            let e = self
                                .error_slot
                                .try_get_error()
                                .expect("cannot enter failed state without setting error slot");
                            let _ = tx.send(Err(e));
                            // Continue until we've errored all queued reqs
                            continue;
                        }
                        None => return,
                    }
                }
            }
        }
    }
/// Consume this `Connection` to form a spawnable future containing its event loop.
Consume this `Connection` to form a spawnable future containing its event loop.
Consume this ` Connection ` to form a spawnable future containing its event loop .
pub async fn run<Rx>(mut self, mut peer_rx: Rx)
    where
        Rx: Stream<Item = Result<Message, SerializationError>> + Unpin,
    {
        // At a high level, the event loop we want is as follows: we check for any
        // incoming messages from the remote peer, check if they should be interpreted
        // as a response to a pending client request, and if not, interpret them as a
        // request from the remote peer to our node.
        //
        // We also need to handle those client requests in the first place. The client
        // requests are received from the corresponding `peer::Client` over a bounded
        // channel (with bound 1, to minimize buffering), but there is no relationship
        // between the stream of client requests and the stream of peer messages, so we
        // cannot ignore one kind while waiting on the other. Moreover, we cannot accept
        // a second client request while the first one is still pending.
        //
        // To do this, we inspect the current request state.
        //
        // If there is no pending request, we wait on either an incoming peer message or
        // an incoming request, whichever comes first.
        //
        // If there is a pending request, we wait only on an incoming peer message, and
        // check whether it can be interpreted as a response to the pending request.
        loop {
            match self.state {
                State::AwaitingRequest => {
                    trace!("awaiting client request or peer message");
                    // CORRECTNESS
                    //
                    // Currently, select prefers the first future if multiple
                    // futures are ready.
                    //
                    // The peer can starve client requests if it sends an
                    // uninterrupted series of messages. But this is unlikely in
                    // practice, due to network delays.
                    //
                    // If both futures are ready, there's no particular reason
                    // to prefer one over the other.
                    //
                    // TODO: use `futures::select!`, which chooses a ready future
                    //       at random, avoiding starvation
                    //       (To use `select!`, we'll need to map the different
                    //       results to a new enum types.)
                    match future::select(peer_rx.next(), self.client_rx.next()).await {
                        Either::Left((None, _)) => {
                            self.fail_with(PeerError::ConnectionClosed);
                        }
                        Either::Left((Some(Err(e)), _)) => self.fail_with(e),
                        Either::Left((Some(Ok(msg)), _)) => {
                            self.handle_message_as_request(msg).await
                        }
                        Either::Right((None, _)) => {
                            trace!("client_rx closed, ending connection");
                            return;
                        }
                        Either::Right((Some(req), _)) => {
                            let span = req.span.clone();
                            self.handle_client_request(req).instrument(span).await
                        }
                    }
                }
                // We're awaiting a response to a client request,
                // so wait on either a peer message, or on a request cancellation.
                State::AwaitingResponse {
                    ref span,
                    ref mut tx,
                    ..
                } => {
                    // we have to get rid of the span reference so we can tamper with the state
                    let span = span.clone();
                    trace!(parent: &span, "awaiting response to client request");
                    let timer_ref = self
                        .request_timer
                        .as_mut()
                        .expect("timeout must be set while awaiting response");
                    // CORRECTNESS
                    //
                    // Currently, select prefers the first future if multiple
                    // futures are ready.
                    //
                    // If multiple futures are ready, we want the cancellation
                    // to take priority, then the timeout, then peer responses.
                    let cancel = future::select(tx.cancellation(), timer_ref);
                    match future::select(cancel, peer_rx.next())
                        .instrument(span.clone())
                        .await
                    {
                        Either::Right((None, _)) => self.fail_with(PeerError::ConnectionClosed),
                        Either::Right((Some(Err(e)), _)) => self.fail_with(e),
                        Either::Right((Some(Ok(peer_msg)), _cancel)) => {
                            // Try to process the message using the handler.
                            // This extremely awkward construction avoids
                            // keeping a live reference to handler across the
                            // call to handle_message_as_request, which takes
                            // &mut self. This is a sign that we don't properly
                            // factor the state required for inbound and
                            // outbound requests.
                            let request_msg = match self.state {
                                State::AwaitingResponse {
                                    ref mut handler, ..
                                } => span.in_scope(|| handler.process_message(peer_msg)),
                                _ => unreachable!("unexpected state after AwaitingResponse: {:?}, peer_msg: {:?}, client_receiver: {:?}",
                                                  self.state,
                                                  peer_msg,
                                                  self.client_rx,
                                ),
                            };
                            // If the message was not consumed, check whether it
                            // should be handled as a request.
                            if let Some(msg) = request_msg {
                                // do NOT instrument with the request span, this is
                                // independent work
                                self.handle_message_as_request(msg).await;
                            } else {
                                // Otherwise, check whether the handler is finished
                                // processing messages and update the state.
                                self.state = match self.state {
                                    State::AwaitingResponse {
                                        handler: Handler::Finished(response),
                                        tx,
                                        ..
                                    } => {
                                        let _ = tx.send(response.map_err(Into::into));
                                        State::AwaitingRequest
                                    }
                                    pending @ State::AwaitingResponse { .. } => pending,
                                    _ => unreachable!(
                                        "unexpected failed connection state while AwaitingResponse: client_receiver: {:?}",
                                        self.client_rx
                                    ),
                                };
                            }
                        }
                        Either::Left((Either::Right(_), _peer_fut)) => {
                            trace!(parent: &span, "client request timed out");
                            let e = PeerError::ClientRequestTimeout;
                            self.state = match self.state {
                                // Special case: ping timeouts fail the connection.
                                State::AwaitingResponse {
                                    handler: Handler::Ping(_),
                                    ..
                                } => {
                                    self.fail_with(e);
                                    State::Failed
                                }
                                // Other request timeouts fail the request.
                                State::AwaitingResponse { tx, .. } => {
                                    let _ = tx.send(Err(e.into()));
                                    State::AwaitingRequest
                                }
                                _ => unreachable!(
                                    "unexpected failed connection state while AwaitingResponse: client_receiver: {:?}",
                                    self.client_rx
                                ),
                            };
                        }
                        Either::Left((Either::Left(_), _peer_fut)) => {
                            trace!(parent: &span, "client request was cancelled");
                            self.state = State::AwaitingRequest;
                        }
                    }
                }
                // We've failed, but we need to flush all pending client
                // requests before we can return and complete the future.
                State::Failed => {
                    match self.client_rx.next().await {
                        Some(InProgressClientRequest { tx, span, .. }) => {
                            trace!(
                                parent: &span,
                                "sending an error response to a pending request on a failed connection"
                            );
                            let e = self
                                .error_slot
                                .try_get_error()
                                .expect("cannot enter failed state without setting error slot");
                            let _ = tx.send(Err(e));
                            // Continue until we've errored all queued reqs
                            continue;
                        }
                        None => return,
                    }
                }
            }
        }
    }
pub async fn run < Rx > ( mut self , mut peer_rx : Rx ) where Rx : Stream < Item = Result < Message , SerializationError > > + Unpin , { // At a high level, the event loop we want is as follows: we check for any // incoming messages from the remote peer, check if they should be interpreted // as a response to a pending client request, and if not, interpret them as a // request from the remote peer to our node. // // We also need to handle those client requests in the first place. The client // requests are received from the corresponding `peer::Client` over a bounded // channel (with bound 1, to minimize buffering), but there is no relationship // between the stream of client requests and the stream of peer messages, so we // cannot ignore one kind while waiting on the other. Moreover, we cannot accept // a second client request while the first one is still pending. // // To do this, we inspect the current request state. // // If there is no pending request, we wait on either an incoming peer message or // an incoming request, whichever comes first. // // If there is a pending request, we wait only on an incoming peer message, and // check whether it can be interpreted as a response to the pending request. loop { match self . state { State :: AwaitingRequest => { trace ! ( " " ) ; // CORRECTNESS // // Currently, select prefers the first future if multiple // futures are ready. // // The peer can starve client requests if it sends an // uninterrupted series of messages. But this is unlikely in // practice, due to network delays. // // If both futures are ready, there's no particular reason // to prefer one over the other. // // TODO: use `futures::select!`, which chooses a ready future //       at random, avoiding starvation //       (To use `select!`, we'll need to map the different //       results to a new enum types.) match future :: select ( peer_rx . next ( ) , self . client_rx . next ( ) ) . await { Either :: Left ( ( None , _ ) ) => { self . fail_with ( PeerError :: ConnectionClosed ) ; } Either :: Left ( ( Some ( Err ( e ) ) , _ ) ) => self . fail_with ( e ) , Either :: Left ( ( Some ( Ok ( msg ) ) , _ ) ) => { self . handle_message_as_request ( msg ) . await } Either :: Right ( ( None , _ ) ) => { trace ! ( " " ) ; return ; } Either :: Right ( ( Some ( req ) , _ ) ) => { let span = req . span . clone ( ) ; self . handle_client_request ( req ) . instrument ( span ) . await } } } // We're awaiting a response to a client request, // so wait on either a peer message, or on a request cancellation. State :: AwaitingResponse { ref span , ref mut tx , .. } => { // we have to get rid of the span reference so we can tamper with the state let span = span . clone ( ) ; trace ! ( parent span " " ) ; let timer_ref = self . request_timer . as_mut ( ) . expect ( " " ) ; // CORRECTNESS // // Currently, select prefers the first future if multiple // futures are ready. // // If multiple futures are ready, we want the cancellation // to take priority, then the timeout, then peer responses. let cancel = future :: select ( tx . cancellation ( ) , timer_ref ) ; match future :: select ( cancel , peer_rx . next ( ) ) . instrument ( span . clone ( ) ) . await { Either :: Right ( ( None , _ ) ) => self . fail_with ( PeerError :: ConnectionClosed ) , Either :: Right ( ( Some ( Err ( e ) ) , _ ) ) => self . fail_with ( e ) , Either :: Right ( ( Some ( Ok ( peer_msg ) ) , _cancel ) ) => { // Try to process the message using the handler. // This extremely awkward construction avoids // keeping a live reference to handler across the // call to handle_message_as_request, which takes // &mut self. This is a sign that we don't properly // factor the state required for inbound and // outbound requests. let request_msg = match self . state { State :: AwaitingResponse { ref mut handler , .. } => span . in_scope ( | | handler . process_message ( peer_msg ) ) , _ => unreachable ! ( " " self state peer_msg self client_rx ) , } ; // If the message was not consumed, check whether it // should be handled as a request. if let Some ( msg ) = request_msg { // do NOT instrument with the request span, this is // independent work self . handle_message_as_request ( msg ) . await ; } else { // Otherwise, check whether the handler is finished // processing messages and update the state. self . state = match self . state { State :: AwaitingResponse { handler : Handler :: Finished ( response ) , tx , .. } => { let _ = tx . send ( response . map_err ( Into :: into ) ) ; State :: AwaitingRequest } pending @ State :: AwaitingResponse { .. } => pending , _ => unreachable ! ( " " self client_rx ) , } ; } } Either :: Left ( ( Either :: Right ( _ ) , _peer_fut ) ) => { trace ! ( parent span " " ) ; let e = PeerError :: ClientRequestTimeout ; self . state = match self . state { // Special case: ping timeouts fail the connection. State :: AwaitingResponse { handler : Handler :: Ping ( _ ) , .. } => { self . fail_with ( e ) ; State :: Failed } // Other request timeouts fail the request. State :: AwaitingResponse { tx , .. } => { let _ = tx . send ( Err ( e . into ( ) ) ) ; State :: AwaitingRequest } _ => unreachable ! ( " " self client_rx ) , } ; } Either :: Left ( ( Either :: Left ( _ ) , _peer_fut ) ) => { trace ! ( parent span " " ) ; self . state = State :: AwaitingRequest ; } } } // We've failed, but we need to flush all pending client // requests before we can return and complete the future. State :: Failed => { match self . client_rx . next ( ) . await { Some ( InProgressClientRequest { tx , span , .. } ) => { trace ! ( parent span " " ) ; let e = self . error_slot . try_get_error ( ) . expect ( " " ) ; let _ = tx . send ( Err ( e ) ) ; // Continue until we've errored all queued reqs continue ; } None => return , } } } } }
Consume this `Connection` to form a spawnable future containing its event loop.
Consume this ` Connection ` to form a spawnable future containing its event loop .
mut self
peer_rx
Rx