pub async fn accept(
        &self,
        tx: Tx,
        rx: Rx,
    ) -> Result<
        (
            Key,
            Option<Chan<S, Sender<Key, Tx, Rx>, Receiver<Key, Tx, Rx>>>,
        ),
        AcceptError<Key, Err>,
    > {
        let (result, ends) = H::over(tx, rx, |chan| (self.handshake)(chan)).await;
        let (resume_kind, key) = match result {
            Ok(key) => key,
            Err(error) => return Err(AcceptError::HandshakeError(error)),
        };
        let (tx, rx) = match ends {
            Ok(ends) => ends,
            Err(_) => return Err(AcceptError::HandshakeIncomplete),
        };

        use dashmap::mapref::entry::Entry;
        match (resume_kind, self.managed.entry(key.clone())) {
            (ResumeKind::Existing, Entry::Vacant(_)) => {
                // If the connection was supposed to be for an existing key but no key
                // existed yet, return an error instead of generating a new channel.
                Err(AcceptError::NoSuchSessionKey(key))
            }
            (ResumeKind::New, Entry::Occupied(_)) => {
                // If the connection was supposed to be for a new key, but the key existed,
                // return an error instead of resuming on that key.
                Err(AcceptError::SessionKeyAlreadyExists(key))
            }
            (ResumeKind::Existing, Entry::Occupied(occupied)) => {
                // If there was already an entry by this key, there's a session, potentially
                // paused, by this key. If the handshake was properly implemented, we're
                // only receiving this tx/rx pair because the other end tried to reconnect,
                // which means that the connection broke in one direction or another, so
                // either we are currently paused, or we're about to discover the issue and
                // pause. Sending the tx/rx will un-pause us and allow us to resume on those
                // fresh channels.
                let send_tx = &occupied.get().send_tx;
                let send_rx = &occupied.get().send_rx;

                use mpsc::error::TrySendError::{Closed, Full};
                if (send_tx.capacity() > 0 || send_tx.is_closed())
                    && (send_rx.capacity() > 0 || send_rx.is_closed())
                {
                    match (send_tx.try_send(tx), send_rx.try_send(rx)) {
                        // This should be impossible, because we have exclusive access due to the
                        // map reference, and nobody else ever sends, so capacity can only increase:
                        (Err(Full(_)), Ok(())) | (Ok(()), Err(Full(_))) => {
                            panic!("partial resumption after capacity check")
                        }
                        // This should also be impossible, but it's less bad if it happens because
                        // if the other side is also full/closed, there's no mismatch that would
                        // lead to the `Sender` having a `tx` that doesn't correspond to the
                        // `Receiver`'s `rx`
                        (Err(Full(_)), Err(Closed(_)))
                        | (Err(Closed(_)), Err(Full(_)))
                        | (Err(Full(_)), Err(Full(_)))
                        // If both are successful, or closed, then everything's fine
                        | (Ok(()), Ok(()))
                        | (Ok(()), Err(Closed(_)))
                        | (Err(Closed(_)), Ok(()))
                        | (Err(Closed(_)), Err(Closed(_))) => Ok((key, None)),
                    }
                } else {
                    Err(AcceptError::NoCapacity)
                }
            }
            (ResumeKind::New, Entry::Vacant(vacant)) => {
                // If there was no entry by this key, then this represents a fresh session,
                // and we should return a channel for the caller to do with what they like
                // (usually, run a session in a separate task).
                let (send_tx, next_tx) = maybe_bounded::channel(self.max_pending_retries);
                let (send_rx, next_rx) = maybe_bounded::channel(self.max_pending_retries);
                let _ = vacant.insert(ConnectionSink {
                    send_tx,
                    send_rx,
                    half_dropped: false.into(),
                });
                let tx = Sender {
                    end: End::new(
                        key.clone(),
                        tx,
                        next_tx,
                        self.recover_tx.clone(),
                        self.managed.clone(),
                        self.timeout,
                    ),
                };
                let rx = Receiver {
                    end: End::new(
                        key.clone(),
                        rx,
                        next_rx,
                        self.recover_rx.clone(),
                        self.managed.clone(),
                        self.timeout,
                    ),
                };
                Ok((key, Some(S::wrap(tx, rx))))
            }
        }
    }
/// Attempt to accept a new connection using this [`Acceptor`].
///
/// If this returns `Ok((key, None))`, the connection has been successfully forwarded to some
/// existing pair of [`Sender`] and [`struct@Receiver`] because the handshake returned a
/// matching key for that session. If this returns `Ok((key, Some(chan)))`, the handshake
/// successfully indicated that a new session should be started for the returned key. Otherwise,
/// an error occurred while trying to accept this connection, and is returned.
Attempt to accept a new connection using this [`Acceptor`].
Attempt to accept a new connection using this [ ` Acceptor ` ] .
pub async fn accept(
        &self,
        tx: Tx,
        rx: Rx,
    ) -> Result<
        (
            Key,
            Option<Chan<S, Sender<Key, Tx, Rx>, Receiver<Key, Tx, Rx>>>,
        ),
        AcceptError<Key, Err>,
    > {
        let (result, ends) = H::over(tx, rx, |chan| (self.handshake)(chan)).await;
        let (resume_kind, key) = match result {
            Ok(key) => key,
            Err(error) => return Err(AcceptError::HandshakeError(error)),
        };
        let (tx, rx) = match ends {
            Ok(ends) => ends,
            Err(_) => return Err(AcceptError::HandshakeIncomplete),
        };
        use dashmap::mapref::entry::Entry;
        match (resume_kind, self.managed.entry(key.clone())) {
            (ResumeKind::Existing, Entry::Vacant(_)) => {
                // If the connection was supposed to be for an existing key but no key
                // existed yet, return an error instead of generating a new channel.
                Err(AcceptError::NoSuchSessionKey(key))
            }
            (ResumeKind::New, Entry::Occupied(_)) => {
                // If the connection was supposed to be for a new key, but the key existed,
                // return an error instead of resuming on that key.
                Err(AcceptError::SessionKeyAlreadyExists(key))
            }
            (ResumeKind::Existing, Entry::Occupied(occupied)) => {
                // If there was already an entry by this key, there's a session, potentially
                // paused, by this key. If the handshake was properly implemented, we're
                // only receiving this tx/rx pair because the other end tried to reconnect,
                // which means that the connection broke in one direction or another, so
                // either we are currently paused, or we're about to discover the issue and
                // pause. Sending the tx/rx will un-pause us and allow us to resume on those
                // fresh channels.
                let send_tx = &occupied.get().send_tx;
                let send_rx = &occupied.get().send_rx;
                use mpsc::error::TrySendError::{Closed, Full};
                if (send_tx.capacity() > 0 || send_tx.is_closed())
                    && (send_rx.capacity() > 0 || send_rx.is_closed())
                {
                    match (send_tx.try_send(tx), send_rx.try_send(rx)) {
                        // This should be impossible, because we have exclusive access due to the
                        // map reference, and nobody else ever sends, so capacity can only increase:
                        (Err(Full(_)), Ok(())) | (Ok(()), Err(Full(_))) => {
                            panic!("partial resumption after capacity check")
                        }
                        // This should also be impossible, but it's less bad if it happens because
                        // if the other side is also full/closed, there's no mismatch that would
                        // lead to the `Sender` having a `tx` that doesn't correspond to the
                        // `Receiver`'s `rx`
                        (Err(Full(_)), Err(Closed(_)))
                        | (Err(Closed(_)), Err(Full(_)))
                        | (Err(Full(_)), Err(Full(_)))
                        // If both are successful, or closed, then everything's fine
                        | (Ok(()), Ok(()))
                        | (Ok(()), Err(Closed(_)))
                        | (Err(Closed(_)), Ok(()))
                        | (Err(Closed(_)), Err(Closed(_))) => Ok((key, None)),
                    }
                } else {
                    Err(AcceptError::NoCapacity)
                }
            }
            (ResumeKind::New, Entry::Vacant(vacant)) => {
                // If there was no entry by this key, then this represents a fresh session,
                // and we should return a channel for the caller to do with what they like
                // (usually, run a session in a separate task).
                let (send_tx, next_tx) = maybe_bounded::channel(self.max_pending_retries);
                let (send_rx, next_rx) = maybe_bounded::channel(self.max_pending_retries);
                let _ = vacant.insert(ConnectionSink {
                    send_tx,
                    send_rx,
                    half_dropped: false.into(),
                });
                let tx = Sender {
                    end: End::new(
                        key.clone(),
                        tx,
                        next_tx,
                        self.recover_tx.clone(),
                        self.managed.clone(),
                        self.timeout,
                    ),
                };
                let rx = Receiver {
                    end: End::new(
                        key.clone(),
                        rx,
                        next_rx,
                        self.recover_rx.clone(),
                        self.managed.clone(),
                        self.timeout,
                    ),
                };
                Ok((key, Some(S::wrap(tx, rx))))
            }
        }
    }
pub async fn accept ( & self , tx : Tx , rx : Rx , ) -> Result < ( Key , Option < Chan < S , Sender < Key , Tx , Rx > , Receiver < Key , Tx , Rx > > > , ) , AcceptError < Key , Err > , > { let ( result , ends ) = H :: over ( tx , rx , | chan | ( self . handshake ) ( chan ) ) . await ; let ( resume_kind , key ) = match result { Ok ( key ) => key , Err ( error ) => return Err ( AcceptError :: HandshakeError ( error ) ) , } ; let ( tx , rx ) = match ends { Ok ( ends ) => ends , Err ( _ ) => return Err ( AcceptError :: HandshakeIncomplete ) , } ; use dashmap :: mapref :: entry :: Entry ; match ( resume_kind , self . managed . entry ( key . clone ( ) ) ) { ( ResumeKind :: Existing , Entry :: Vacant ( _ ) ) => { // If the connection was supposed to be for an existing key but no key // existed yet, return an error instead of generating a new channel. Err ( AcceptError :: NoSuchSessionKey ( key ) ) } ( ResumeKind :: New , Entry :: Occupied ( _ ) ) => { // If the connection was supposed to be for a new key, but the key existed, // return an error instead of resuming on that key. Err ( AcceptError :: SessionKeyAlreadyExists ( key ) ) } ( ResumeKind :: Existing , Entry :: Occupied ( occupied ) ) => { // If there was already an entry by this key, there's a session, potentially // paused, by this key. If the handshake was properly implemented, we're // only receiving this tx/rx pair because the other end tried to reconnect, // which means that the connection broke in one direction or another, so // either we are currently paused, or we're about to discover the issue and // pause. Sending the tx/rx will un-pause us and allow us to resume on those // fresh channels. let send_tx = & occupied . get ( ) . send_tx ; let send_rx = & occupied . get ( ) . send_rx ; use mpsc :: error :: TrySendError :: { Closed , Full } ; if ( send_tx . capacity ( ) > 0 || send_tx . is_closed ( ) ) && ( send_rx . capacity ( ) > 0 || send_rx . is_closed ( ) ) { match ( send_tx . try_send ( tx ) , send_rx . try_send ( rx ) ) { // This should be impossible, because we have exclusive access due to the // map reference, and nobody else ever sends, so capacity can only increase: ( Err ( Full ( _ ) ) , Ok ( ( ) ) ) | ( Ok ( ( ) ) , Err ( Full ( _ ) ) ) => { panic ! ( " " ) } // This should also be impossible, but it's less bad if it happens because // if the other side is also full/closed, there's no mismatch that would // lead to the `Sender` having a `tx` that doesn't correspond to the // `Receiver`'s `rx` ( Err ( Full ( _ ) ) , Err ( Closed ( _ ) ) ) | ( Err ( Closed ( _ ) ) , Err ( Full ( _ ) ) ) | ( Err ( Full ( _ ) ) , Err ( Full ( _ ) ) ) // If both are successful, or closed, then everything's fine | ( Ok ( ( ) ) , Ok ( ( ) ) ) | ( Ok ( ( ) ) , Err ( Closed ( _ ) ) ) | ( Err ( Closed ( _ ) ) , Ok ( ( ) ) ) | ( Err ( Closed ( _ ) ) , Err ( Closed ( _ ) ) ) => Ok ( ( key , None ) ) , } } else { Err ( AcceptError :: NoCapacity ) } } ( ResumeKind :: New , Entry :: Vacant ( vacant ) ) => { // If there was no entry by this key, then this represents a fresh session, // and we should return a channel for the caller to do with what they like // (usually, run a session in a separate task). let ( send_tx , next_tx ) = maybe_bounded :: channel ( self . max_pending_retries ) ; let ( send_rx , next_rx ) = maybe_bounded :: channel ( self . max_pending_retries ) ; let _ = vacant . insert ( ConnectionSink { send_tx , send_rx , half_dropped : false . into ( ) , } ) ; let tx = Sender { end : End :: new ( key . clone ( ) , tx , next_tx , self . recover_tx . clone ( ) , self . managed . clone ( ) , self . timeout , ) , } ; let rx = Receiver { end : End :: new ( key . clone ( ) , rx , next_rx , self . recover_rx . clone ( ) , self . managed . clone ( ) , self . timeout , ) , } ; Ok ( ( key , Some ( S :: wrap ( tx , rx ) ) ) ) } } }
Attempt to accept a new connection using this [`Acceptor`].
Attempt to accept a new connection using this [ ` Acceptor ` ] .
&self
tx
Tx
rx
Rx