pub async fn run<F>(mut self, mut init_server: F)
    where
        F: FnMut(P, Sender<SessionError>) -> (Sender<R>, Receiver<R>),
    {
        // NOTE: The following comment illustrates the original synchronous version, but the gist of it is the same.
        // In a loop, add the `rx` of `demux`, and all the `rxs` of `mux` to a `Select`, see which one is ready:
        // * If `error.rx` returns `Ok` then a protocol violation occurred and we can exit the loop.
        // * If `demux.rx` returns `Ok` then get or create the protocol in both `mux` and `demux`, and dispatch into the corresponding channel in `demux.txs`.
        // * If `demux.rx` returns `Err` then the connection is closed and we can exit the loop.
        // * If any of `mux.rxs` return `Ok` then wrap the message and send to `mux.tx`.
        // * If any of `mux.rxs` return `Err` then that protocol is finished and can be removed from both `mux` and `demux`.

        // We have to declare a `FuturesUnordered` here, can't store it in a field becuase it has an opaque type, `impl Future`.
        let mut mux = FuturesUnordered::new();

        loop {
            select! {
                // One of the sessions detection a violation of some rule and now all of them need to be closed.
                // This will never return `None` because we never close the error channel.
                Some(_) = self.errors.rx.recv() => break,

                // One of the session handlers has a message to send to the client.
                // This might return `None` when there are no sessions, but the pattern will discard this branch
                // and wait on the others. The first incoming message will then establish a handler and the next
                // loop should see it matching.
                Some((pid, outgoing, rx)) = mux.next() => {
                    match outgoing {
                        Some(msg) => {
                            self.handle_outgoing_reply(pid, msg);
                            // Put the receiver back in the mux, so further replies from the session are handled.
                            mux.push(Self::recv_outgoing(pid, rx));
                        }
                        // The outgoing channel was closed on our side.
                        None => self.remove(pid),
                    }
                },

                // There is an incoming request from the client.
                incoming = self.demux.recv() => {
                    match incoming {
                        Some((pid, msg)) => {
                            if let Some(rx) = self.handle_incoming_request(pid, msg, &mut init_server) {
                                // Register the receiver with the reply multiplexer.
                                mux.push(Self::recv_outgoing(pid, rx));
                            }
                        },
                        // The incoming channel is closed.
                        None => break,
                    }
                },
            }
        }
    }
/// Start consuming messages from the client, creating new protocol handlers
/// where one doesn't exist already.
///
/// We must also listen to an internal channel that signals the abortion
/// of the whole connection due to some protocol violation.
///
/// `init_server` should initiate the server side of the protocol, listening
/// to incoming messages from the client; it should return the channels we
/// can send the client messages to, and receive the replies over. When
/// sending or receiving from these channels fail, the protocol is over
/// and should be removed.
Start consuming messages from the client, creating new protocol handlers
where one doesn't exist already.

We must also listen to an internal channel that signals the abortion
of the whole connection due to some protocol violation.

`init_server` should initiate the server side of the protocol, listening
to incoming messages from the client; it should return the channels we
can send the client messages to, and receive the replies over. When
sending or receiving from these channels fail, the protocol is over
and should be removed.
Start consuming messages from the client creating new protocol handlers where one doesn ' t exist already . We must also listen to an internal channel that signals the abortion of the whole connection due to some protocol violation . ` init_server ` should initiate the server side of the protocol listening to incoming messages from the client ; it should return the channels we can send the client messages to and receive the replies over . When sending or receiving from these channels fail the protocol is over and should be removed .
pub async fn run<F>(mut self, mut init_server: F)
    where
        F: FnMut(P, Sender<SessionError>) -> (Sender<R>, Receiver<R>),
    {
        // NOTE: The following comment illustrates the original synchronous version, but the gist of it is the same.
        // In a loop, add the `rx` of `demux`, and all the `rxs` of `mux` to a `Select`, see which one is ready:
        // * If `error.rx` returns `Ok` then a protocol violation occurred and we can exit the loop.
        // * If `demux.rx` returns `Ok` then get or create the protocol in both `mux` and `demux`, and dispatch into the corresponding channel in `demux.txs`.
        // * If `demux.rx` returns `Err` then the connection is closed and we can exit the loop.
        // * If any of `mux.rxs` return `Ok` then wrap the message and send to `mux.tx`.
        // * If any of `mux.rxs` return `Err` then that protocol is finished and can be removed from both `mux` and `demux`.
        // We have to declare a `FuturesUnordered` here, can't store it in a field becuase it has an opaque type, `impl Future`.
        let mut mux = FuturesUnordered::new();
        loop {
            select! {
                // One of the sessions detection a violation of some rule and now all of them need to be closed.
                // This will never return `None` because we never close the error channel.
                Some(_) = self.errors.rx.recv() => break,
                // One of the session handlers has a message to send to the client.
                // This might return `None` when there are no sessions, but the pattern will discard this branch
                // and wait on the others. The first incoming message will then establish a handler and the next
                // loop should see it matching.
                Some((pid, outgoing, rx)) = mux.next() => {
                    match outgoing {
                        Some(msg) => {
                            self.handle_outgoing_reply(pid, msg);
                            // Put the receiver back in the mux, so further replies from the session are handled.
                            mux.push(Self::recv_outgoing(pid, rx));
                        }
                        // The outgoing channel was closed on our side.
                        None => self.remove(pid),
                    }
                },
                // There is an incoming request from the client.
                incoming = self.demux.recv() => {
                    match incoming {
                        Some((pid, msg)) => {
                            if let Some(rx) = self.handle_incoming_request(pid, msg, &mut init_server) {
                                // Register the receiver with the reply multiplexer.
                                mux.push(Self::recv_outgoing(pid, rx));
                            }
                        },
                        // The incoming channel is closed.
                        None => break,
                    }
                },
            }
        }
    }
pub async fn run < F > ( mut self , mut init_server : F ) where F : FnMut ( P , Sender < SessionError > ) -> ( Sender < R > , Receiver < R > ) , { // NOTE: The following comment illustrates the original synchronous version, but the gist of it is the same. // In a loop, add the `rx` of `demux`, and all the `rxs` of `mux` to a `Select`, see which one is ready: // * If `error.rx` returns `Ok` then a protocol violation occurred and we can exit the loop. // * If `demux.rx` returns `Ok` then get or create the protocol in both `mux` and `demux`, and dispatch into the corresponding channel in `demux.txs`. // * If `demux.rx` returns `Err` then the connection is closed and we can exit the loop. // * If any of `mux.rxs` return `Ok` then wrap the message and send to `mux.tx`. // * If any of `mux.rxs` return `Err` then that protocol is finished and can be removed from both `mux` and `demux`. // We have to declare a `FuturesUnordered` here, can't store it in a field becuase it has an opaque type, `impl Future`. let mut mux = FuturesUnordered :: new ( ) ; loop { select ! { // One of the sessions detection a violation of some rule and now all of them need to be closed. // This will never return `None` because we never close the error channel. Some ( ) self errors rx recv ( ) break // One of the session handlers has a message to send to the client. // This might return `None` when there are no sessions, but the pattern will discard this branch // and wait on the others. The first incoming message will then establish a handler and the next // loop should see it matching. Some ( ( pid outgoing rx ) ) mux next ( ) { match outgoing { Some ( msg ) { self handle_outgoing_reply ( pid msg ) // Put the receiver back in the mux, so further replies from the session are handled. mux push ( Self recv_outgoing ( pid rx ) ) } // The outgoing channel was closed on our side. None self remove ( pid ) } } // There is an incoming request from the client. incoming self demux recv ( ) { match incoming { Some ( ( pid msg ) ) { if let Some ( rx ) self handle_incoming_request ( pid msg mut init_server ) { // Register the receiver with the reply multiplexer. mux push ( Self recv_outgoing ( pid rx ) ) } } // The incoming channel is closed. None break } } } } }
Start consuming messages from the client, creating new protocol handlers
where one doesn't exist already.
Start consuming messages from the client creating new protocol handlers where one doesn ' t exist already .
mut self
init_server
F