pub(crate) async fn apply(
        mut self,
        db: &Db,
        dst: &mut Connection,
        shutdown: &mut Shutdown,
    ) -> crate::Result<()> {
        // Each individual channel subscription is handled using a
        // `sync::broadcast` channel. Messages are then fanned out to all
        // clients currently subscribed to the channels.
        //
        // An individual client may subscribe to multiple channels and may
        // dynamically add and remove channels from its subscription set. To
        // handle this, a `StreamMap` is used to track active subscriptions. The
        // `StreamMap` merges messages from individual broadcast channels as
        // they are received.
        let mut subscriptions = StreamMap::new();

        loop {
            // `self.channels` is used to track additional channels to subscribe
            // to. When new `SUBSCRIBE` commands are received during the
            // execution of `apply`, the new channels are pushed onto this vec.
            for channel_name in self.channels.drain(..) {
                subscribe_to_channel(channel_name, &mut subscriptions, db, dst).await?;
            }

            // Wait for one of the following to happen:
            //
            // - Receive a message from one of the subscribed channels.
            // - Receive a subscribe or unsubscribe command from the client.
            // - A server shutdown signal.
            select! {
                // Receive messages from subscribed channels
                Some((channel_name, msg)) = subscriptions.next() => {
                    dst.write_frame(&make_message_frame(channel_name, msg)).await?;
                },
                res = dst.read_frame() => {
                    let frame = match res? {
                        Some(frame) => frame,
                        // This happens if the remote client has disconnected.
                        None => return Ok(())
                    };

                    handle_command(
                        frame,
                        &mut self.channels,
                        &mut subscriptions,
                        dst,
                    ).await?;
                },
                _ = shutdown.recv() => {
                    return Ok(());
                }
            };
        }
    }
/// Apply the `Subscribe` command to the specified `Db` instance.
///
/// This function is the entry point and includes the initial list of
/// channels to subscribe to. Additional `subscribe` and `unsubscribe`
/// commands may be received from the client and the list of subscriptions
/// are updated accordingly.
///
/// [here]: https://redis.io/topics/pubsub
Apply the `Subscribe` command to the specified `Db` instance.
This function is the entry point and includes the initial list of
channels to subscribe to. Additional `subscribe` and `unsubscribe`
commands may be received from the client and the list of subscriptions
are updated accordingly.


Apply the ` Subscribe ` command to the specified ` Db ` instance . This function is the entry point and includes the initial list of channels to subscribe to . Additional ` subscribe ` and ` unsubscribe ` commands may be received from the client and the list of subscriptions are updated accordingly .
pub(crate) async fn apply(
        mut self,
        db: &Db,
        dst: &mut Connection,
        shutdown: &mut Shutdown,
    ) -> crate::Result<()> {
        // Each individual channel subscription is handled using a
        // `sync::broadcast` channel. Messages are then fanned out to all
        // clients currently subscribed to the channels.
        //
        // An individual client may subscribe to multiple channels and may
        // dynamically add and remove channels from its subscription set. To
        // handle this, a `StreamMap` is used to track active subscriptions. The
        // `StreamMap` merges messages from individual broadcast channels as
        // they are received.
        let mut subscriptions = StreamMap::new();
        loop {
            // `self.channels` is used to track additional channels to subscribe
            // to. When new `SUBSCRIBE` commands are received during the
            // execution of `apply`, the new channels are pushed onto this vec.
            for channel_name in self.channels.drain(..) {
                subscribe_to_channel(channel_name, &mut subscriptions, db, dst).await?;
            }
            // Wait for one of the following to happen:
            //
            // - Receive a message from one of the subscribed channels.
            // - Receive a subscribe or unsubscribe command from the client.
            // - A server shutdown signal.
            select! {
                // Receive messages from subscribed channels
                Some((channel_name, msg)) = subscriptions.next() => {
                    dst.write_frame(&make_message_frame(channel_name, msg)).await?;
                },
                res = dst.read_frame() => {
                    let frame = match res? {
                        Some(frame) => frame,
                        // This happens if the remote client has disconnected.
                        None => return Ok(())
                    };
                    handle_command(
                        frame,
                        &mut self.channels,
                        &mut subscriptions,
                        dst,
                    ).await?;
                },
                _ = shutdown.recv() => {
                    return Ok(());
                }
            };
        }
    }
pub ( crate ) async fn apply ( mut self , db : & Db , dst : & mut Connection , shutdown : & mut Shutdown , ) -> crate :: Result < ( ) > { // Each individual channel subscription is handled using a // `sync::broadcast` channel. Messages are then fanned out to all // clients currently subscribed to the channels. // // An individual client may subscribe to multiple channels and may // dynamically add and remove channels from its subscription set. To // handle this, a `StreamMap` is used to track active subscriptions. The // `StreamMap` merges messages from individual broadcast channels as // they are received. let mut subscriptions = StreamMap :: new ( ) ; loop { // `self.channels` is used to track additional channels to subscribe // to. When new `SUBSCRIBE` commands are received during the // execution of `apply`, the new channels are pushed onto this vec. for channel_name in self . channels . drain ( .. ) { subscribe_to_channel ( channel_name , & mut subscriptions , db , dst ) . await ? ; } // Wait for one of the following to happen: // // - Receive a message from one of the subscribed channels. // - Receive a subscribe or unsubscribe command from the client. // - A server shutdown signal. select ! { // Receive messages from subscribed channels Some ( ( channel_name msg ) ) subscriptions next ( ) { dst write_frame ( make_message_frame ( channel_name msg ) ) await } res dst read_frame ( ) { let frame match res { Some ( frame ) frame // This happens if the remote client has disconnected. None return Ok ( ( ) ) } handle_command ( frame mut self channels mut subscriptions dst ) await } shutdown recv ( ) { return Ok ( ( ) ) } } ; } }
Apply the `Subscribe` command to the specified `Db` instance.
Apply the ` Subscribe ` command to the specified ` Db ` instance .
mut self
db
&Db
dst
&mut Connection
shutdown
&mut Shutdown