pub(crate) fn poll_event(&mut self, cx: &mut Context<'_>) -> Poll<ConnectionEvent> {
        // Nothing more can be done if the connection is drained.
        // Return `Pending` without registering the waker, essentially freezing the task forever.
        if self.connection.is_drained() {
            tracing::error!("poll_event called on a drained connection");
            return Poll::Pending;
        }

        // Process events that the endpoint has sent to us.
        loop {
            match Pin::new(&mut self.from_endpoint).poll_next(cx) {
                Poll::Ready(Some(event)) => self.connection.handle_event(event),
                Poll::Ready(None) => {
                    assert!(self.closed.is_none());
                    let err = Error::ClosedChannel;
                    self.closed = Some(err.clone());
                    return Poll::Ready(ConnectionEvent::ConnectionLost(err));
                }
                Poll::Pending => break,
            }
        }

        'send_pending: loop {
            // Sending the pending event to the endpoint. If the endpoint is too busy, we just
            // stop the processing here.
            // There is a bit of a question in play here: should be continue to accept events
            // through `from_endpoint` if `to_endpoint` is busy?
            // We need to be careful to avoid a potential deadlock if both `from_endpoint` and
            // `to_endpoint` are full. As such, we continue to transfer data from `from_endpoint`
            // to the `quinn_proto::Connection` (see above).
            // However we don't deliver substream-related events to the user as long as
            // `to_endpoint` is full. This should propagate the back-pressure of `to_endpoint`
            // being full to the user.
            if let Some(pending_to_endpoint) = &mut self.pending_to_endpoint {
                match Future::poll(Pin::new(pending_to_endpoint), cx) {
                    Poll::Pending => return Poll::Pending,
                    Poll::Ready(()) => self.pending_to_endpoint = None,
                }
            }

            let now = Instant::now();

            // Poll the connection for packets to send on the UDP socket and try to send them on
            // `to_endpoint`.
            while let Some(transmit) = self.connection.poll_transmit(now) {
                let endpoint = self.endpoint.clone();
                assert!(self.pending_to_endpoint.is_none());
                self.pending_to_endpoint = Some(Box::pin(async move {
                    // TODO: ECN bits not handled
                    endpoint
                        .send_udp_packet(transmit.destination, transmit.contents)
                        .await;
                }));
                continue 'send_pending;
            }

            // The connection also needs to be able to send control messages to the endpoint. This is
            // handled here, and we try to send them on `to_endpoint` as well.
            while let Some(endpoint_event) = self.connection.poll_endpoint_events() {
                let endpoint = self.endpoint.clone();
                let connection_id = self.connection_id;
                assert!(self.pending_to_endpoint.is_none());
                self.pending_to_endpoint = Some(Box::pin(async move {
                    endpoint
                        .report_quinn_event(connection_id, endpoint_event)
                        .await;
                }));
                continue 'send_pending;
            }

            // Timeout system.
            // We break out of the following loop until if `poll_timeout()` returns `None` or if
            // polling `self.next_timeout` returns `Poll::Pending`.
            loop {
                if let Some(next_timeout) = &mut self.next_timeout {
                    match Future::poll(Pin::new(next_timeout), cx) {
                        Poll::Ready(()) => {
                            self.connection.handle_timeout(now);
                            self.next_timeout = None;
                        }
                        Poll::Pending => break,
                    }
                } else if let Some(when) = self.connection.poll_timeout() {
                    if when <= now {
                        self.connection.handle_timeout(now);
                    } else {
                        let delay = when - now;
                        self.next_timeout = Some(futures_timer::Delay::new(delay));
                    }
                } else {
                    break;
                }
            }

            // The final step consists in handling the events related to the various substreams.
            while let Some(event) = self.connection.poll() {
                match event {
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Opened {
                        dir: quinn_proto::Dir::Uni,
                    })
                    | quinn_proto::Event::Stream(quinn_proto::StreamEvent::Available {
                        dir: quinn_proto::Dir::Uni,
                    })
                    | quinn_proto::Event::DatagramReceived => {
                        // We don't use datagrams or unidirectional streams. If these events
                        // happen, it is by some code not compatible with libp2p-quic.
                        self.connection
                            .close(Instant::now(), From::from(0u32), Default::default());
                    }
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Readable { id }) => {
                        return Poll::Ready(ConnectionEvent::StreamReadable(id));
                    }
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Writable { id }) => {
                        return Poll::Ready(ConnectionEvent::StreamWritable(id));
                    }
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Available {
                        dir: quinn_proto::Dir::Bi,
                    }) => {
                        return Poll::Ready(ConnectionEvent::StreamAvailable);
                    }
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Opened {
                        dir: quinn_proto::Dir::Bi,
                    }) => {
                        return Poll::Ready(ConnectionEvent::StreamOpened);
                    }
                    quinn_proto::Event::ConnectionLost { reason } => {
                        assert!(self.closed.is_none());
                        self.is_handshaking = false;
                        let err = Error::Quinn(reason);
                        self.closed = Some(err.clone());
                        return Poll::Ready(ConnectionEvent::ConnectionLost(err));
                    }
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Finished {
                        id,
                        stop_reason: _,
                    }) => {
                        // TODO: transmit `stop_reason`
                        return Poll::Ready(ConnectionEvent::StreamFinished(id));
                    }
                    quinn_proto::Event::Connected => {
                        assert!(self.is_handshaking);
                        assert!(!self.connection.is_handshaking());
                        self.is_handshaking = false;
                        return Poll::Ready(ConnectionEvent::Connected);
                    }
                }
            }

            break;
        }

        Poll::Pending
    }
/// Polls the connection for an event that happend on it.
Polls the connection for an event that happend on it.
Polls the connection for an event that happend on it .
pub(crate) fn poll_event(&mut self, cx: &mut Context<'_>) -> Poll<ConnectionEvent> {
        // Nothing more can be done if the connection is drained.
        // Return `Pending` without registering the waker, essentially freezing the task forever.
        if self.connection.is_drained() {
            tracing::error!("poll_event called on a drained connection");
            return Poll::Pending;
        }
        // Process events that the endpoint has sent to us.
        loop {
            match Pin::new(&mut self.from_endpoint).poll_next(cx) {
                Poll::Ready(Some(event)) => self.connection.handle_event(event),
                Poll::Ready(None) => {
                    assert!(self.closed.is_none());
                    let err = Error::ClosedChannel;
                    self.closed = Some(err.clone());
                    return Poll::Ready(ConnectionEvent::ConnectionLost(err));
                }
                Poll::Pending => break,
            }
        }
        'send_pending: loop {
            // Sending the pending event to the endpoint. If the endpoint is too busy, we just
            // stop the processing here.
            // There is a bit of a question in play here: should be continue to accept events
            // through `from_endpoint` if `to_endpoint` is busy?
            // We need to be careful to avoid a potential deadlock if both `from_endpoint` and
            // `to_endpoint` are full. As such, we continue to transfer data from `from_endpoint`
            // to the `quinn_proto::Connection` (see above).
            // However we don't deliver substream-related events to the user as long as
            // `to_endpoint` is full. This should propagate the back-pressure of `to_endpoint`
            // being full to the user.
            if let Some(pending_to_endpoint) = &mut self.pending_to_endpoint {
                match Future::poll(Pin::new(pending_to_endpoint), cx) {
                    Poll::Pending => return Poll::Pending,
                    Poll::Ready(()) => self.pending_to_endpoint = None,
                }
            }
            let now = Instant::now();
            // Poll the connection for packets to send on the UDP socket and try to send them on
            // `to_endpoint`.
            while let Some(transmit) = self.connection.poll_transmit(now) {
                let endpoint = self.endpoint.clone();
                assert!(self.pending_to_endpoint.is_none());
                self.pending_to_endpoint = Some(Box::pin(async move {
                    // TODO: ECN bits not handled
                    endpoint
                        .send_udp_packet(transmit.destination, transmit.contents)
                        .await;
                }));
                continue 'send_pending;
            }
            // The connection also needs to be able to send control messages to the endpoint. This is
            // handled here, and we try to send them on `to_endpoint` as well.
            while let Some(endpoint_event) = self.connection.poll_endpoint_events() {
                let endpoint = self.endpoint.clone();
                let connection_id = self.connection_id;
                assert!(self.pending_to_endpoint.is_none());
                self.pending_to_endpoint = Some(Box::pin(async move {
                    endpoint
                        .report_quinn_event(connection_id, endpoint_event)
                        .await;
                }));
                continue 'send_pending;
            }
            // Timeout system.
            // We break out of the following loop until if `poll_timeout()` returns `None` or if
            // polling `self.next_timeout` returns `Poll::Pending`.
            loop {
                if let Some(next_timeout) = &mut self.next_timeout {
                    match Future::poll(Pin::new(next_timeout), cx) {
                        Poll::Ready(()) => {
                            self.connection.handle_timeout(now);
                            self.next_timeout = None;
                        }
                        Poll::Pending => break,
                    }
                } else if let Some(when) = self.connection.poll_timeout() {
                    if when <= now {
                        self.connection.handle_timeout(now);
                    } else {
                        let delay = when - now;
                        self.next_timeout = Some(futures_timer::Delay::new(delay));
                    }
                } else {
                    break;
                }
            }
            // The final step consists in handling the events related to the various substreams.
            while let Some(event) = self.connection.poll() {
                match event {
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Opened {
                        dir: quinn_proto::Dir::Uni,
                    })
                    | quinn_proto::Event::Stream(quinn_proto::StreamEvent::Available {
                        dir: quinn_proto::Dir::Uni,
                    })
                    | quinn_proto::Event::DatagramReceived => {
                        // We don't use datagrams or unidirectional streams. If these events
                        // happen, it is by some code not compatible with libp2p-quic.
                        self.connection
                            .close(Instant::now(), From::from(0u32), Default::default());
                    }
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Readable { id }) => {
                        return Poll::Ready(ConnectionEvent::StreamReadable(id));
                    }
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Writable { id }) => {
                        return Poll::Ready(ConnectionEvent::StreamWritable(id));
                    }
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Available {
                        dir: quinn_proto::Dir::Bi,
                    }) => {
                        return Poll::Ready(ConnectionEvent::StreamAvailable);
                    }
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Opened {
                        dir: quinn_proto::Dir::Bi,
                    }) => {
                        return Poll::Ready(ConnectionEvent::StreamOpened);
                    }
                    quinn_proto::Event::ConnectionLost { reason } => {
                        assert!(self.closed.is_none());
                        self.is_handshaking = false;
                        let err = Error::Quinn(reason);
                        self.closed = Some(err.clone());
                        return Poll::Ready(ConnectionEvent::ConnectionLost(err));
                    }
                    quinn_proto::Event::Stream(quinn_proto::StreamEvent::Finished {
                        id,
                        stop_reason: _,
                    }) => {
                        // TODO: transmit `stop_reason`
                        return Poll::Ready(ConnectionEvent::StreamFinished(id));
                    }
                    quinn_proto::Event::Connected => {
                        assert!(self.is_handshaking);
                        assert!(!self.connection.is_handshaking());
                        self.is_handshaking = false;
                        return Poll::Ready(ConnectionEvent::Connected);
                    }
                }
            }
            break;
        }
        Poll::Pending
    }
pub ( crate ) fn poll_event ( & mut self , cx : & mut Context < ' _ > ) -> Poll < ConnectionEvent > { // Nothing more can be done if the connection is drained. // Return `Pending` without registering the waker, essentially freezing the task forever. if self . connection . is_drained ( ) { tracing :: error ! ( " " ) ; return Poll :: Pending ; } // Process events that the endpoint has sent to us. loop { match Pin :: new ( & mut self . from_endpoint ) . poll_next ( cx ) { Poll :: Ready ( Some ( event ) ) => self . connection . handle_event ( event ) , Poll :: Ready ( None ) => { assert ! ( self closed is_none ( ) ) ; let err = Error :: ClosedChannel ; self . closed = Some ( err . clone ( ) ) ; return Poll :: Ready ( ConnectionEvent :: ConnectionLost ( err ) ) ; } Poll :: Pending => break , } } ' send_pending : loop { // Sending the pending event to the endpoint. If the endpoint is too busy, we just // stop the processing here. // There is a bit of a question in play here: should be continue to accept events // through `from_endpoint` if `to_endpoint` is busy? // We need to be careful to avoid a potential deadlock if both `from_endpoint` and // `to_endpoint` are full. As such, we continue to transfer data from `from_endpoint` // to the `quinn_proto::Connection` (see above). // However we don't deliver substream-related events to the user as long as // `to_endpoint` is full. This should propagate the back-pressure of `to_endpoint` // being full to the user. if let Some ( pending_to_endpoint ) = & mut self . pending_to_endpoint { match Future :: poll ( Pin :: new ( pending_to_endpoint ) , cx ) { Poll :: Pending => return Poll :: Pending , Poll :: Ready ( ( ) ) => self . pending_to_endpoint = None , } } let now = Instant :: now ( ) ; // Poll the connection for packets to send on the UDP socket and try to send them on // `to_endpoint`. while let Some ( transmit ) = self . connection . poll_transmit ( now ) { let endpoint = self . endpoint . clone ( ) ; assert ! ( self pending_to_endpoint is_none ( ) ) ; self . pending_to_endpoint = Some ( Box :: pin ( async move { // TODO: ECN bits not handled endpoint . send_udp_packet ( transmit . destination , transmit . contents ) . await ; } ) ) ; continue ' send_pending ; } // The connection also needs to be able to send control messages to the endpoint. This is // handled here, and we try to send them on `to_endpoint` as well. while let Some ( endpoint_event ) = self . connection . poll_endpoint_events ( ) { let endpoint = self . endpoint . clone ( ) ; let connection_id = self . connection_id ; assert ! ( self pending_to_endpoint is_none ( ) ) ; self . pending_to_endpoint = Some ( Box :: pin ( async move { endpoint . report_quinn_event ( connection_id , endpoint_event ) . await ; } ) ) ; continue ' send_pending ; } // Timeout system. // We break out of the following loop until if `poll_timeout()` returns `None` or if // polling `self.next_timeout` returns `Poll::Pending`. loop { if let Some ( next_timeout ) = & mut self . next_timeout { match Future :: poll ( Pin :: new ( next_timeout ) , cx ) { Poll :: Ready ( ( ) ) => { self . connection . handle_timeout ( now ) ; self . next_timeout = None ; } Poll :: Pending => break , } } else if let Some ( when ) = self . connection . poll_timeout ( ) { if when <= now { self . connection . handle_timeout ( now ) ; } else { let delay = when - now ; self . next_timeout = Some ( futures_timer :: Delay :: new ( delay ) ) ; } } else { break ; } } // The final step consists in handling the events related to the various substreams. while let Some ( event ) = self . connection . poll ( ) { match event { quinn_proto :: Event :: Stream ( quinn_proto :: StreamEvent :: Opened { dir : quinn_proto :: Dir :: Uni , } ) | quinn_proto :: Event :: Stream ( quinn_proto :: StreamEvent :: Available { dir : quinn_proto :: Dir :: Uni , } ) | quinn_proto :: Event :: DatagramReceived => { // We don't use datagrams or unidirectional streams. If these events // happen, it is by some code not compatible with libp2p-quic. self . connection . close ( Instant :: now ( ) , From :: from ( 0u32 ) , Default :: default ( ) ) ; } quinn_proto :: Event :: Stream ( quinn_proto :: StreamEvent :: Readable { id } ) => { return Poll :: Ready ( ConnectionEvent :: StreamReadable ( id ) ) ; } quinn_proto :: Event :: Stream ( quinn_proto :: StreamEvent :: Writable { id } ) => { return Poll :: Ready ( ConnectionEvent :: StreamWritable ( id ) ) ; } quinn_proto :: Event :: Stream ( quinn_proto :: StreamEvent :: Available { dir : quinn_proto :: Dir :: Bi , } ) => { return Poll :: Ready ( ConnectionEvent :: StreamAvailable ) ; } quinn_proto :: Event :: Stream ( quinn_proto :: StreamEvent :: Opened { dir : quinn_proto :: Dir :: Bi , } ) => { return Poll :: Ready ( ConnectionEvent :: StreamOpened ) ; } quinn_proto :: Event :: ConnectionLost { reason } => { assert ! ( self closed is_none ( ) ) ; self . is_handshaking = false ; let err = Error :: Quinn ( reason ) ; self . closed = Some ( err . clone ( ) ) ; return Poll :: Ready ( ConnectionEvent :: ConnectionLost ( err ) ) ; } quinn_proto :: Event :: Stream ( quinn_proto :: StreamEvent :: Finished { id , stop_reason : _ , } ) => { // TODO: transmit `stop_reason` return Poll :: Ready ( ConnectionEvent :: StreamFinished ( id ) ) ; } quinn_proto :: Event :: Connected => { assert ! ( self is_handshaking ) ; assert ! ( self connection is_handshaking ( ) ) ; self . is_handshaking = false ; return Poll :: Ready ( ConnectionEvent :: Connected ) ; } } } break ; } Poll :: Pending }
Polls the connection for an event that happend on it.
Polls the connection for an event that happend on it .
&mut self
cx
&mut Context<'_>
&mut self
cx
&mut Context<'_>