# Mini-Redis教程

## 准备工作

###  获取代码

```shell
git clone --recursive https://github.com/DragonOS-Community/rust_camp_code.git
```

教程版请切换curse-mini-redis分支

###  运行项目

####  运行服务端

```shell
RUST_LOG=debug cargo run --bin mini-redis-server
```

运行服务端之后，就可以运行redis操作的代码和redis的命令了。详情请参考example文件夹下的代码和redis的命令。

####  运行example的代码

```shell
cargo run --example hello_world
```

####  运行redis命令

```shell
cargo run --bin mini-redis-cli set foo bar

cargo run --bin mini-redis-cli get foo
```

##  Module

###  client

> 最小的redis会话。为所支持的命令提供异步连接。

代码位于src/clients/client.rs

####  Client

#####  Struct

用于建立和存储客户端连接

```rust
pub struct Client{
    /// 客户端连接
    connection:Connection,
}
```

#####  Implementation

```rust
impl Client{
    /// 建立连接
    pub async fn connect<T: ToSocketAddrs>(addr: T) -> crate::Result<Client> {
        // 与addr建立tcp连接并获取套接字
        let socket = TcpStream::connect(addr).await?;
        // 使用套接字建立连接
        let connection = Connection::new(socket);
        // 返回客户端
        Ok(Client { connection })
    }

    /// ping客户端
    #[instrument(skip(self))]
    pub async fn ping(&mut self, msg: Option<Bytes>) -> crate::Result<Bytes> {
        // 构建ping消息
        let frame = Ping::new(msg).into_frame();
        // 日志记录ping
        debug!(request = ?frame);
        // 发送ping消息
        self.connection.write_frame(&frame).await?;
        // 读取响应，对Frame::Simple、Frame::Bulk、error类型进行处理
        match self.read_response().await? {
            Frame::Simple(value) => Ok(value.into()),
            Frame::Bulk(value) => Ok(value),
            frame => Err(frame.to_error()),
        }
    }

    /// 获取key对应的value
    #[instrument(skip(self))]
    pub async fn get(&mut self,key:&str) -> Result<Option<Bytes>> {
        // 创建获取'key'的值的'Get'命令,并将其转换成frame格式
        let frame = Get::new(key).into_frame();
        // 日志记录frme
        debug!(request = ?frame);

        // 将frame通过连接异步写给套接字
        self.connection.write_frame(&frame).await?;

        // 异步读响应,并匹配Simple、Bulk、NUll的情况
        match self.read_response().await? {
            Frame::Simple(value) => Ok(Some(value.into())),
            Frame::Bulk(value) => Ok(Some(value)),
            Frame::Null => Ok(None),
            frame => Err(frame.to_error()),
        }
    }

    /// 设置key对应的value
    #[instrument(skip(self))]
    pub async fn set(&mut self, key: &str, value: Bytes) -> crate::Result<()> {
        // 通过set_cmd异步创建过期时间为None的Set命令
        self.set_cmd(Set::new(key, value, None)).await
    }

    /// 设置key对应的value，value在expiration后过期
    #[instrument(skip(self))]
    pub async fn set_expires(
        &mut self,
        key: &str,
        value: Bytes,
        expiration: Duration) -> crate::Result<()>{
        // 通过set_cmd异步创建过期时间为expiration的Set命令
        self.set_cmd(Set::new(key, value, Some(expiration))).await
    }

    /// Set命令的主要逻辑
    async fn set_cmd(&mut self, cmd: Set) -> crate::Result<()> {
        // 将cmd转化为frame的形式
        let frame = cmd.into_frame();
        // 将frame写入连接
        self.connection.write_frame(&frame).await?;
        // 读取对frame的响应，并匹配响应状态
        // 执行成功响应为"OK"
        // 其他响应均为失败
        match self.read_response().await? {
            Frame::Simple(response) if response == "OK" => Ok(()),
            frame => Err(frame.to_error()),
        }    
    }

    /// 将信息推送给指定的频道
    #[instrument(skip(self))]
    pub async fn publish(&mut self, channel: &str, message: Bytes) -> crate::Result<u64> {
        // 将Publish命令转成frame的格式
        let frame = Publish::new(channel, message).into_frame();

        // 日志记录frme
        debug!(request = ?frame);

        // 将frame通过连接异步写给套接字
        self.connection.write_frame(&frame).await?;

        // 异步读请求，并匹配请求类型
        // 如果是整数帧，则返回OK(response)
        // 否则返回错误
        match self.read_response().await? {
            Frame::Integer(response) => Ok(response),
            frame => Err(frame.to_error()),
        }
    }

    /// SUBSCRIBE的主要逻辑
    async fn subscribe_cmd(&mut self, channels: &[String]) -> crate::Result<()> {
        // 将Subscribe命令转成frame的格式
        let frame = Subscribe::new(channels.to_vec()).into_frame();

        // 日志记录frme
        debug!(request = ?frame);

        // 将frame通过连接异步写给套接字
        self.connection.write_frame(&frame).await?;

        // 遍历channels，服务端通过响应确认订阅每个频道
        for channel in channels {
            // 获取响应
            let response = self.read_response().await?;

            // 分析不同情况的响应
            match response {
                Frame::Array(ref frame) => match frame.as_slice() {
                    [subscribe, schannel, ..]
                        if *subscribe == "subscribe" && *schannel == channel => {}
                    _ => return Err(response.to_error()),
                },
                frame => return Err(frame.to_error()),
            };
        }

        Ok(())
    }

    /// 监听若干个指定的频道
    #[instrument(skip(self))]
    pub async fn subscribe(mut self, channels: Vec<String>) -> crate::Result<Subscriber> {
        // 异步调用subscribe_cmd，客户端状态将会转为subscriber
        self.subscribe_cmd(&channels).await?;

        // 返回Subscriber类型对象
        Ok(Subscriber {
            client: self,
            subscribed_channels: channels,
        })
    }

    /// 读取响应
    async fn read_response(&mut self) -> crate::Result<Frame> {
        // 异步读取连接的frame
        let response = self.connection.read_frame().await?;
        // 日志记录读取信息
        debug!(?response);

        // 解析响应，判断响应的类型
        // Some(Frame::Error(msg))
        // Some(frame)
        // None：响应为None意味着服务端关闭
        match response {
            // Error frames are converted to `Err`
            Some(Frame::Error(msg)) => Err(msg.into()),
            Some(frame) => Ok(frame),
            None => {
                // Receiving `None` here indicates the server has closed the
                // connection without sending a frame. This is unexpected and is
                // represented as a "connection reset by peer" error.
                let err = Error::new(ErrorKind::ConnectionReset, "connection reset by server");

                Err(err.into())
            }
        }

    }
}
```

####  Message

#####  Struct

```rust
/// A message received on a subscribed channel.
#[derive(Debug, Clone)]
pub struct Message {
    pub channel: String,
    pub content: Bytes,
}
```

####  Subscriber

> 订阅者类型，在客户端订阅若干频道后会转变成订阅者，此时只能执行publish/subscribe相关的命令

代码位置：src/clients/client.rs

#####  Struct

```rust
pub struct Subscriber {
    /// 客户端
    client: Client,
    /// 所订阅的频道集合
    subscribed_channels: Vec<String>,
}
```

#####  Implementation

```rust
impl Subscriber{
    /// 获取订阅的频道集合
    pub fn get_subscribed(&self) -> &[String] {
        &self.subscribed_channels
    }

    /// 获取客户端收到的下一条消息（可能需要等待）
    pub async fn next_message(&mut self) -> crate::Result<Option<Message>> {
        match self.client.connection.read_frame().await? {
            Some(mframe) => {
                // 使用日志记录mframe
                debug!(?mframe);
                match mframe {
                    Frame::Array(ref frame) => match frame.as_slice() {
                      // frame分片后格式为[message,channel,content]
                      // 当接收到信息时，message == 'message'
                      // 成立则将信息转化为Message的形式返回
                      // 否则返回错误
                      [message, channel, content] if *message == "message" => Ok(Some(Message {
                            channel: channel.to_string(),
                            content: Bytes::from(content.to_string()),
                        })),
                        _ => Err(mframe.to_error()),
                    },
                    frame => Err(frame.to_error()),
                }
            }
            None => Ok(None),
        }
    }

    /// 将收到的message转换成stream形式
    pub fn into_stream(mut self) -> impl Stream<Item = crate::Result<Message>> {
        try_stream! {
            while let Some(message) = self.next_message().await? {
                yield message;
            }
        }
    }

    /// 订阅新的频道集合
    #[instrument(skip(self))]
    pub async fn subscribe(&mut self, channels: &[String]) -> crate::Result<()> {
        // 调用subscribe_cmd，订阅新的频道集合
        self.client.subscribe_cmd(channels).await?;

        // 更新当前的已订阅频道
        self.subscribed_channels
            .extend(channels.iter().map(Clone::clone));

        Ok(())
    }

    /// 取消订阅指定的频道
    #[instrument(skip(self))]
    pub async fn unsubscribe(&mut self, channels: &[String]) -> crate::Result<()> {
        // 创建Unsubcribe命令，并转为frame形式
        let frame = Unsubscribe::new(channels).into_frame();
        // 日志记录请求
        debug!(request = ?frame);

        // 将请求写入连接
        self.client.connection.write_frame(&frame).await?;

        // 如果channels为空，那么取消订阅所有频道
        // 否则只取消channels中指定的频道
        let num = if channels.is_empty() {
            self.subscribed_channels.len()
        } else {
            channels.len()
        };

        // 解析响应
        for _ in 0..num {
            // 读取响应
            todo!();
            // 判断响应类型
            match response {
                Frame::Array(ref frame) => match frame.as_slice() {
                    // frame分片后格式为['unsubscribe',channel,..]
                    // 判断是否取消订阅
                    [unsubscribe, channel, ..] if *unsubscribe == "unsubscribe" => {
                       // 获取当前订阅数组的长度
                        let len = self.subscribed_channels.len();
                        // 长度为0，返回错误
                        if len == 0 {
                            // There must be at least one channel
                            return Err(response.to_error());
                        }
                        // 当channel存在于subscribed_channels时将其删除
                        self.subscribed_channels.retain(|c| *channel != &c[..]);

                        // 删除数大于1则返回错误
                        if self.subscribed_channels.len() != len - 1 {
                            return Err(response.to_error());
                        }                    }
                    _ => return Err(response.to_error()),
                },
                frame => return Err(frame.to_error()),
            };
        }

        Ok(())
    }   
}
```

###  blocking_client

> 最小的阻塞性Redis客户端

代码位置：src/clients/blocking_client.rs

####  BlockingClient

#####  Struct

```rust
pub struct BlockingClient{
     /// 异步的客户端
     inner: crate::clients::Client,
     /// 运行时环境，负责调度和管理异步任务的执行
     rt: Runtime,
}
```

#####  Implementation

```rust
impl BlockingClient{
    /// 建立连接
    pub fn connect<T: ToSocketAddrs>(addr: T) -> crate::Result<BlockingClient> {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()?;

        let inner = rt.block_on(crate::clients::Client::connect(addr))?;

        Ok(BlockingClient { inner, rt })
    }    

    /// 异步获取指定值
    pub fn get(&mut self, key: &str) -> crate::Result<Option<Bytes>> {
        // 异步调用get获取key对应的值
        self.rt.block_on(self.inner.get(key))
    }

    /// 异步给key赋值
    pub fn set(&mut self, key: &str, value: Bytes) -> crate::Result<()> {
        // 异步调用set给key赋值
        self.rt.block_on(self.inner.set(key, value))
    }    

   /// 异步给key赋值，并指定过期时间
       pub fn set_expires(
        &mut self,
        key: &str,
        value: Bytes,
        expiration: Duration,
    ) -> crate::Result<()> {
        // 异步调用set_expires给key赋值
        self.rt
            .block_on(self.inner.set_expires(key, value, expiration))    }

    /// 异步推送消息
    pub fn publish(&mut self, channel: &str, message: Bytes) -> crate::Result<u64> {
        // 异步调用publish给指定channel推送消息
        self.rt.block_on(self.inner.publish(channel, message))
    }

    /// 异步执行订阅指定频道操作，并将Client转换为BlockingSubcriber状态
    pub fn subscribe(self, channels: Vec<String>) -> crate::Result<BlockingSubscriber> {
        // 异步调用subscribe函数，转换client状态
        let subscriber = self.rt.block_on(self.inner.subscribe(channels))?;
        // 返回BlockingSubscriber
        Ok(BlockingSubscriber {
            inner: subscriber,
            rt: self.rt,
        })
    }    
}
```

####  BlockingSubscriber

#####  Struct

```rust
pub struct BlockingSubscriber {
    /// 异步的订阅者
    inner: crate::clients::Subscriber,
    /// 运行时环境，负责调度和管理异步任务的执行
    rt: Runtime,
}
```

#####  Implementation

```rust
impl BlockingSubscriber{
    /// 获取订阅的channel集合
    pub fn get_subscribed(&self) -> &[String] {
        // 获取订阅的channel集合，并返回
        self.inner.get_subscribed()
    }

       /// 异步接收下一条信息
       pub fn next_message(&mut self) -> crate::Result<Option<Message>> {
        // 异步执行获取下一条信息的操作
        self.rt.block_on(self.inner.next_message())
    }

    /// 异步订阅新的频道集合
    pub fn subscribe(&mut self, channels: &[String]) -> crate::Result<()> {
        // 异步订阅新的频道集合
        self.rt.block_on(self.inner.subscribe(channels))
    }

    /// 异步执行取消订阅指定频道集操作
    pub fn unsubscribe(&mut self, channels: &[String]) -> crate::Result<()> {
        // 异步执行取消订阅指定频道集操作
        self.rt.block_on(self.inner.unsubscribe(channels))
    }

    /// 获取BlockingSubscriber的迭代器
    pub fn into_iter(self) -> impl Iterator<Item = crate::Result<Message>> {
        SubscriberIterator {
            inner: self.inner,
            rt: self.rt,
        }
    }
}
```

####  SubscriberIterator

#####  Struct

```rust
/// BlockingSubscriber的迭代器
/// 通过`Subscriber::into_iter`可以获取
struct SubscriberIterator {
    /// BlockingSubscriber中的Subscriber
    inner: crate::clients::Subscriber,

    /// BlockingSubscriber中的Runtime
    rt: Runtime,
}
```

#####  Implementation

```rust
impl Iterator for SubscriberIterator {
    /// 定义迭代器的元素类型
    type Item = crate::Result<Message>;

    /// 获取迭代器的下一个元素
    /// 返回Some(Ok(message)) 表示存在下一个元素
    /// 返回Some(Err(error)) 表示获取下一个元素出错
    /// 返回None 表示没有下一个元素
    fn next(&mut self) -> Option<crate::Result<Message>>
    {
        self.rt.block_on(self.inner.next_message()).transpose()
    }
}
```

###  cmd

####  Get

> 获取key对应的value

- 如果key不存在，返回nil
- 如果value不是string类型，返回error

代码位置：src/cmd/get.rs

#####  Struct

```rust
#[derive(Debug)]
pub struct Get {
    /// Name of the key to get
    key: String,
}
```

#####  Implementation

```rust
impl Get{
    /// 创建一个Get命令对象
    pub fn new(key: impl ToString) -> Get {
        Get {
            key: key.to_string(),
        }
    }

    /// Get the key
    pub fn key(&self) -> &str {
        &self.key
    }

    /// 解析frame，并获取对应Get命令
    pub(crate) fn parse_frames(parse: &mut Parse) -> crate::Result<Get> {
        let key = parse.next_string()?;
        Ok(Get { key })
    }

    /// 对数据库执行Get命令
    #[instrument(skip(self, db, dst))]
    pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> crate::Result<()> {
        // 从数据库中获取value
        // 如果值存在，则将其转成Frame::Bulk形式
        // 否则转成Frame::Null
        let response = if let Some(value) = db.get(&self.key) {
            // If a value is present, it is written to the client in "bulk"
            // format.
            Frame::Bulk(value)
        } else {
            // If there is no value, `Null` is written.
            Frame::Null
        };

        // 日志记录响应
        debug!(?response);

        // 将响应写入连接
        dst.write_frame(&response).await?;

        Ok(())
    }

    /// 将命令转成Frame格式，在客户端将格式化的Get
    ///命令发给服务端时会调用。
    pub(crate) fn into_frame(self) -> Frame {
        let mut frame = Frame::array();
        /// 将命令以["get",key]的形式放入frame。注意每个元素都要转成字节形式
        todo!();
        frame
    }
}
```

####  Publish

代码位置：src/cmd/publish.rs

#####  Struct

```rust
/// 向指定频道推送消息
#[derive(Debug)]
pub struct Publish {
    /// 频道名
    channel: String,

    /// 要发送的消息
    message: Bytes,
}
```

#####  Implementation

```rust
impl Publish{
    /// 创建Publish命令
    pub(crate) fn new(channel: impl ToString, message: Bytes) -> Publish {
        Publish {
            channel: channel.to_string(),
            message,
        }
    }

    /// 解析frame，并获取对应Publish命令
    pub(crate) fn parse_frames(parse: &mut Parse) -> crate::Result<Publish> {
        // 获取string形式的channel
        let channel = parse.next_string()?;

        // 获取字节形式的message
        let message = parse.next_bytes()?;

        // 返回对应Publish
        Ok(Publish { channel, message })
    }

    /// 对数据库执行publish命令
    pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> crate::Result<()>{
        // 向数据库连接执行publish操作，并获取到订阅channel的subscriber的数量
        let num_subscribers = db.publish(&self.channel, self.message);

        // 将subscriber_num转换成整数帧，并将其写入连接
        let response = Frame::Integer(num_subscribers as u64);
        dst.write_frame(&response).await?;

        Ok(())
    }

    /// 将publish命令转成frame的形式
    pub(crate) fn into_frame(self) -> Frame {
        let mut frame = Frame::array();
        // 将publish命令以["publish",channel,message]的形式写入数组。注意每个元素都要转成字节形式
        frame.push_bulk(Bytes::from("publish".as_bytes()));
        frame.push_bulk(Bytes::from(self.channel.into_bytes()));
        frame.push_bulk(self.message);

        frame
    }
}
```

####  Set

代码位置：src/cmd/set.rs

#####  Struct

```rust
/// 设置key的值
#[derive(Debug)]
pub struct Set {
    /// the lookup key
    key: String,

    /// the value to be stored
    value: Bytes,

    /// When to expire the key
    expire: Option<Duration>,
}
```

#####  Implementation

```rust
impl Set{
    /// 创建一个set命令
    /// 
    /// 如果不需要过期时间则expire为None
    pub fn new(key: impl ToString, value: Bytes, expire: Option<Duration>) -> Set {
        Set {
            key: key.to_string(),
            value,
            expire,
        }
    }

    /// Get the key
    pub fn key(&self) -> &str {
        &self.key
    }

    /// Get the value
    pub fn value(&self) -> &Bytes {
        &self.value
    }

    /// Get the expire
    pub fn expire(&self) -> Option<Duration> {
        self.expire
    }

    /// 解析frame
    pub(crate) fn parse_frames(parse: &mut Parse) -> crate::Result<Set> {
        use ParseError::EndOfStream;

        // 使用 parse.next_string() 方法读取下一个字符串字段作为键（key）。
        let key = parse.next_string()?;

        // 使用 parse.next_bytes() 方法读取下一个字节字段作为值（value）。
        let value = parse.next_bytes()?;

        // 创建一个mut expire，用于存储过期时间。初始值为 None。
        let mut expire = None;

        // 分析其他情况的帧
        match parse.next_string() {
            Ok(s) if s.to_uppercase() == "EX" => {
                // 如果下一个字符串字段是 "EX"（不区分大小写），则表示设置了过期时间且单位是secs
                // 下一个值应该是一个整数，并转成Duration类型。
                let secs = parse.next_int()?;
                expire = Some(Duration::from_secs(secs));
            }
            Ok(s) if s.to_uppercase() == "PX" => {
                // 如果下一个字符串字段是 "PX"（不区分大小写），则表示设置了过期时间且单位是millis
                // 下一个值应该是一个整数，并转成Duration类型。
                let ms = parse.next_int()?;
                expire = Some(Duration::from_millis(ms));
            }
            // 如果下一个字符串字段不是 "EX" 或 "PX"，则表示设置了不支持的选项，将返回一个错误，指示 SET 目前仅支持过期时间选项。
            Ok(_) => return Err("currently `SET` only supports the expiration option".into()),
            // 如果解析到达流的末尾，表示没有进一步的数据需要解析，这是正常的运行时情况，表示没有指定 SET 的选项。
            Err(EndOfStream) => {}
            // 其他所有错误都会返回一个错误，导致连接被终止。
            Err(err) => return Err(err.into()),
        }

        Ok(Set { key, value, expire })
    }

    /// 对数据库连接执行set命令
    #[instrument(skip(self, db, dst))]
    pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> crate::Result<()> {
        // 执行set
        db.set(self.key, self.value, self.expire);

        // 创建执行成功响应，并写给连接
        let response = Frame::Simple("OK".to_string());
        debug!(?response);
        dst.write_frame(&response).await?;

        Ok(())
    }

    /// 将set命令转成frame形式
    pub(crate) fn into_frame(self) -> Frame {
        let mut frame = Frame::array();
        // 将命令以["set",key,value]的形式写入frame
        frame.push_bulk(Bytes::from("set".as_bytes()));
        frame.push_bulk(Bytes::from(self.key.into_bytes()));
        frame.push_bulk(self.value);
        // 判断是否有过期时间，如果有过期时间以["px",ms]的格式写入frame
        if let Some(ms) = self.expire {
            // 可以用以下两种方式设置过期时间，在这里使用第二种，因为精度更高
            // 1. SET key value EX seconds
            // 2. SET key value PX milliseconds
            frame.push_bulk(Bytes::from("px".as_bytes()));
            frame.push_int(ms.as_millis() as u64);
        }
        frame
    }
}
```

####  Subscribe+Message

代码位置：src/cmd/subscribe.rs

#####  Struct

```rust
/// 订阅命令
/// 
/// 一旦客户端进入订阅状态，它只能提出 
/// SUBSCRIBE, PSUBSCRIBE, UNSUBSCRIBE,
/// PUNSUBSCRIBE, PING 和 QUIT命令
#[derive(Debug)]
pub struct Subscribe {
    channels: Vec<String>,
}

/// 信息流
type Messages = Pin<Box<dyn Stream<Item = Bytes> + Send>>;
```

#####  Implementation

```rust
impl Subscribe{
    /// 创建一个新的订阅者以监听频道
    pub(crate) fn new(channels: &[String]) -> Subscribe {
        Subscribe {
            channels: channels.to_vec(),
        }
    }
     /// 解析subscribe帧
     pub(crate) fn parse_frames(parse: &mut Parse) -> crate::Result<Subscribe> {
        use ParseError::EndOfStream;
        // 获取第一个channel，并判断是否为none，不为none则写入数组
        let mut channels = vec![parse.next_string()?];
        // 循环解析剩下的channel，并将其写入数组
        loop {
            match parse.next_string() {
                // 推送消息给频道
                Ok(s) => channels.push(s),
                // EndOfStream说明不会再有消息了
                Err(EndOfStream) => break,
                // 错误报告
                Err(err) => return Err(err.into()),
            }
        }
        // 返回结果
        Ok(Subscribe { channels })
    }

    /// 数据库执行subscribe命令
    pub(crate) async fn apply(
        mut self,
        db: &Db,
        dst: &mut Connection,
        shutdown: &mut Shutdown,
    ) -> crate::Result<()> {
        // 创建一个 StreamMap 对象，用于跟踪活动的订阅。使用map可以防止订阅重复
        let mut subscriptions = StreamMap::new();

        loop {
            // 遍历channels，drain(..)会将管道从channels中移除
            // subscribe_to_channel会订阅传入的管道
            // await等待管道订阅成功
            for channel_name in self.channels.drain(..) {
                subscribe_to_channel(channel_name, &mut subscriptions, db, dst).await?;
            }

            // 使用 select! 宏等待以下几种情况中的任何一种发生：
            // 1. 从已订阅的通道接收到消息。
            // 2. 从客户端接收到订阅或取消订阅命令。
            // 3. 收到服务器关闭信号。
            select! {
                // 接收来自已订阅通道的消息。
                // subscriptions.next() 返回一个异步迭代器
                // 当有新的消息到达时，会返回 Some((channel_name, msg))
                // 其中 channel_name 是通道名称，msg 是消息内容。
                Some((channel_name, msg)) = subscriptions.next() => {
                    // 将收到的消息构造成消息帧写给客户端
                    dst.write_frame(&make_message_frame(channel_name, msg)).await?;
                }
                // dst.read_frame() 返回一个异步操作结果，
                //当有新的帧可用时，会返回 Some(frame)，其中 frame 是读取到的帧。如果客户端断开连接，会返回 None。
                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(());
                }
            };
        }
    }

    /// 将命令转为frame类型
    pub(crate) fn into_frame(self) -> Frame {
        let mut frame = Frame::array();
        // 将命令以["subscribe",channel1,channel2...]的形
        //式放入数组
        frame.push_bulk(Bytes::from("unsubscribe".as_bytes()));
        for channel in self.channels {
            frame.push_bulk(Bytes::from(channel.into_bytes()));
        }
        frame
    }



}
```

#####  Function

```rust
async fn subscribe_to_channel(
    channel_name: String,
    subscriptions: &mut StreamMap<String, Messages>,
    db: &Db,
    dst: &mut Connection,
) -> crate::Result<()> {
    // 向数据发送订阅请求
    let mut rx = db.subscribe(channel_name.clone());

    // 接受订阅回复
    let rx = Box::pin(async_stream::stream! {
        loop {
            match rx.recv().await {
                Ok(msg) => yield msg,
                // If we lagged in consuming messages, just resume.
                Err(broadcast::error::RecvError::Lagged(_)) => {}
                Err(_) => break,
            }
        }
    });

    // 将订阅成功的管道加入到订阅map中
    subscriptions.insert(channel_name.clone(), rx);

    // 响应成功订阅报文
    let response = make_subscribe_frame(channel_name, subscriptions.len());
    dst.write_frame(&response).await?;

    Ok(())
}

/// 处理subscribe和unsubscibe的逻辑
async fn handle_command(
    frame: Frame,
    subscribe_to: &mut Vec<String>,
    subscriptions: &mut StreamMap<String, Messages>,
    dst: &mut Connection,
) -> crate::Result<()> {
    // 匹配命令，判断要执行的操作
    match Command::from_frame(frame)? {
        Command::Subscribe(subscribe) => {
        // 将要订阅的频道加入到订阅map          
            subscribe_to.extend(subscribe.channels.into_iter());
        }
        Command::Unsubscribe(mut unsubscribe) => {
            // 判断是否指定了要取消订阅的频道
            // 如果没有则需要订阅所有channels
            if unsubscribe.channels.is_empty() {
                unsubscribe.channels = subscriptions
                    .keys()
                    .map(|channel_name| channel_name.to_string())
                    .collect();
            }
            // 将需要取消订阅的频道从map中移除
            for channel_name in unsubscribe.channels {
                  subscriptions.remove(&channel_name);
                  // 生成unsubscribe响应
                  let response = make_unsubscribe_frame(channel_name, subscriptions.len());
                  // 写入unsubscribe响应
                  dst.write_frame(&response).await?;
            }
        }
        command => {
            // 其他命令都看作是unknown命令
            let cmd = Unknown::new(command.get_name());
            cmd.apply(dst).await?;
        }
    }
    Ok(())
}
```

####  Unsubscribe

代码位置：src/cmd/subcribe.rs

#####  Struct

```rust
/// 取消订阅命令
///
/// 如果channels为空，则取消订阅所有频道
#[derive(Clone, Debug)]
pub struct Unsubscribe {
    channels: Vec<String>,
}
```

#####  Implementation

```rust
impl Unsubscribe{
     pub(crate) fn new(channels: &[String]) -> Unsubscribe {
        Unsubscribe {
            channels: channels.to_vec(),
        }
    }
    /// 解析unsubscribe帧
    pub(crate) fn parse_frames(parse: &mut Parse) -> Result<Unsubscribe, ParseError> {
        use ParseError::EndOfStream;

        // 创建数组，用于存储channels
        let mut channels = vec![];

        // 循环解析数据，将channel放入数组
        loop {
            match parse.next_string() {
                // 将数据推送到要取消订阅的频道列表
                Ok(s) => channels.push(s),
                // EndOfStream代表数据解析完成
                Err(EndOfStream) => break,
                // 其他错误
                Err(err) => return Err(err),
            }
        }

        // 返回命令
    }
    /// 将unsubscribe命令转成帧
    pub(crate) fn into_frame(self) -> Frame {
        let mut frame = Frame::array();
        // 将命令以["unsubscribe",channel1,channel2...]形
        // 式放入数组
        frame.push_bulk(Bytes::from("unsubscribe".as_bytes()));

        for channel in self.channels {
            frame.push_bulk(Bytes::from(channel.into_bytes()));
        }
        frame
    }
}
```

####  Unknown

#####  Struct

```rust
#[derive(Debug)]
pub struct Unknown {
    command_name: String,
}
```

#####  Implementation

```rust
impl Unknown{
    /// 创建unknown命令
    pub(crate) fn new(key: impl ToString) -> Unknown {
        Unknown {
            command_name: key.to_string(),
        }
    }

    /// 获取命令名
    pub(crate) fn get_name(&self) -> &str {
        &self.command_name
    }

    /// 回应客户端不支持当前命令
    #[instrument(skip(self, dst))]
    pub(crate) async fn apply(self, dst: &mut Connection) -> crate::Result<()> {
        // 生成错误帧响应
        let response = Frame::Error(format!("ERR unknown command '{}'", self.command_name));
        // 日志记录响应
        debug!(?response);
        // 将帧写入连接
        dst.write_frame(&response).await?;
        Ok(())
    }
}
```

####  Ping

#####  Struct

```rust
#[derive(Debug, Default)]
pub struct Ping {
    /// 自定义消息返回
    msg: Option<Bytes>,
}
```

#####  Implement

```rust
impl Ping {
    /// 创建一个新的携带自定信息的Ping命令
    pub fn new(msg: Option<Bytes>) -> Ping {
        Ping { msg }
    }

    /// 解析ping命令
    pub(crate) fn parse_frames(parse: &mut Parse) -> crate::Result<Ping> {
        match parse.next_bytes() {
            Ok(msg) => Ok(Ping::new(Some(msg))),
            Err(ParseError::EndOfStream) => Ok(Ping::default()),
            Err(e) => Err(e.into()),
        }
    }

    /// ping命令执行逻辑
    ///
    /// 响应会被写入dst。
    /// 服务端受到命令并执行时会执行。
    #[instrument(skip(self, dst))]
    pub(crate) async fn apply(self, dst: &mut Connection) -> crate::Result<()> {
        let response = match self.msg {
            None => Frame::Simple("PONG".to_string()),
            Some(msg) => Frame::Bulk(msg),
        };

        debug!(?response);

        dst.write_frame(&response).await?;

        Ok(())
    }

    /// 将命令转换成frame
    ///This is called by the client when encoding a `Ping` command to send
    /// to the server.
    /// 当客户端需要发送ping命令给服务端时会被调用
    pub(crate) fn into_frame(self) -> Frame {
        let mut frame = Frame::array();
        frame.push_bulk(Bytes::from("ping".as_bytes()));
        if let Some(msg) = self.msg {
            frame.push_bulk(msg);
        }
        frame
    }
}
```

###  frame

> redis协议帧

####  enum

```rust
/// A frame in the Redis protocol.
#[derive(Clone, Debug)]
pub enum Frame {
    Simple(String),
    Error(String),
    Integer(u64),
    Bulk(Bytes),
    Null,
    Array(Vec<Frame>),
}

#[derive(Debug)]
pub enum Error {
    /// Not enough data is available to parse a message
    Incomplete,

    /// Invalid message encoding
    Other(crate::Error),
}
```

###  sever

> mini-redis的服务端。提供了一个异步的`run`函数，用于监听连接，并为每个连接生成一个任务

####  Listener

#####  Struct

```rust
/// 服务端用于监听的连接
#[derive(Debug)]
struct Listener {
    /// 可共享的数据库连接句柄（Arc形式）
    db_holder: DbDropGuard,

    /// TCP监听类，由run函数提供
    listener: TcpListener,

    /// 信号量，用于限制连接数
    limit_connections: Arc<Semaphore>,

    /// 用于广播shutdown信号
    notify_shutdown: broadcast::Sender<()>,

    /// 用于在客户端完成所有任务后安全关闭连接。
    /// 当一个连接被初始化，它会保存‵shutdown_complete_tx‵的clone，被关闭时回收
    /// 当所有listener的连接都被关闭后，sender也会被释放
    /// 所有任务被完成后，`shutdown_complete_rx.recv()`会返回`None`
    shutdown_complete_tx: mpsc::Sender<()>,
}
```

#####  Implementation

```rust
impl Listener {
    /// 启动server监听服务
    async fn run(&mut self) -> crate::Result<()> {
        info!("accepting inbound connections");

        loop {
            // 等待有名额创建连接
            let permit = self
                .limit_connections
                .clone()
                .acquire_owned()
                .await
                .unwrap();

            // 建立tcp连接，获取套接字
            let socket = self.accept().await?;

            // 存储连接状态
            let mut handler = Handler {
                // 获取数据库连接
                db: self.db_holder.db(),

                // 新建连接
                connection: Connection::new(socket),

                // 用于接受shutdown信号
                shutdown: Shutdown::new(self.notify_shutdown.subscribe()),

                // 一旦所有的clone被释放，会通过此成员来通知
                _shutdown_complete: self.shutdown_complete_tx.clone(),
            };

                // 创建一个异步的任务执行连接要做的操作
                tokio::spawn(async move {
                // 执行连接，并记录错误
                if let Err(err) = handler.run().await {
                    error!(cause = ?err, "connection error");
                }
                // 归还连接名额
                drop(permit);
            });
        }
    }

    /// 建立tcp连接，获取套接字
    /// 如果发生错误，最大重连次数为6，最后一次等待时间为64s。重连流程参考tcp超时重发机制。
    async fn accept(&mut self) -> crate::Result<TcpStream> {
        let mut backoff = 1;

        // 尝试接受连接
        loop {
            // 成功直接返回tcp连接 失败返回错误信息
            match self.listener.accept().await {
                Ok((socket, _)) => return Ok(socket),
                Err(err) => {
                    if backoff > 64 {
                        // 等待时间过长，返回错误
                        return Err(err.into());
                    }
                }
            }

            // 睡眠
            time::sleep(Duration::from_secs(backoff)).await;

            // 指数增长等待时间
            backoff *= 2;
        }
    }
}
```

####  Handler

#####  Struct

```rust
/// 服务端的具体连接
#[derive(Debug)]
struct Handler {
    /// 数据库连接
    db: Db,

    /// tcp连接
    connection: Connection,

    /// 监听shutdown信号
    shutdown: Shutdown,

    /// 不直接使用，用来判断当前类对象是否被释放
    _shutdown_complete: mpsc::Sender<()>,
}
```

#####  Implementation

```rust
impl Handler {
    /// 处理单个连接
    /// 
    /// 当接收到shutdown信号时，等到该连接处于安全状态时，连接会断开
    #[instrument(skip(self))]
    async fn run(&mut self) -> crate::Result<()> {
        // 循环处理帧
        while !self.shutdown.s_shutdown() {
            let maybe_frame = tokio::select! {
                res = self.connection.read_frame() => res?,
                _ = self.shutdown.recv() => {
                    // 接收到shutdown信号，退出run函数会使任务结束
                    return Ok(());
                }
            };

            // 判断从远端收到的帧（maybe_frame）是否有内容，收到‵None`时说明远端关闭
            let frame = match maybe_frame {
                Some(frame) => frame,
                None => return Ok(()),
            };

            // 将frame转为命令形式
            let cmd = Command::from_frame(frame)?;

            // 日志记录接收到的命令
            debug!(?cmd);

            // 异步执行命令
            cmd.apply(&self.db, &mut self.connection, &mut self.shutdown)
                .await?;
        }

        Ok(())
    }
}
```

####  Fn Run

```rust
/// mini-redis的服务端启动函数
pub async fn run(listener: TcpListener, shutdown: impl Future) {
    // 初始化shutdown信号广播管道
    let (notify_shutdown, _) = broadcast::channel(1);
    // 初始化shutdown关闭机制
    let (shutdown_complete_tx, mut shutdown_complete_rx) = mpsc::channel(1);

    // 初始化监听连接
    let mut server = Listener {
        listener,
        db_holder: DbDropGuard::new(),
        limit_connections: Arc::new(Semaphore::new(MAX_CONNECTIONS)),
        notify_shutdown,
        shutdown_complete_tx,
    };

    // 监听是否有连接到达
    tokio::select! {
        res = server.run() => {
            // 处理连接失败的情况
            if let Err(err) = res {
                error!(cause = %err, "failed to accept");
            }
        }
        _ = shutdown => {
            info!("shutting down");
        }
    }

    let Listener {
        shutdown_complete_tx,
        notify_shutdown,
        ..
    } = server;

    // 所有的处在subscribe状态的任务会接收到shutdown信号并且退出
    drop(notify_shutdown);
    // 回收监听者的shutdown_complete_tx，意味着只有其他连接持有shutdown_complete_tx了
    drop(shutdown_complete_tx);

    // 等待所有活跃连接完成其任务
    // 当所有shutdown_complete_tx都被释放，说明所有连接都断开了
    // 此时`recv()`会返回`None`并且`mpsc`管道会被关闭
    let _ = shutdown_complete_rx.recv().await;
}
```

##  Function

###  Fn Buffer

```rust
/// 创建一个请求缓冲区
pub fn buffer(client: Client) -> Buffer {
    // 创建一个容量为32的异步通道。tx为发送端，rx为接收端
    let (tx, rx) = channel(32);

    // 使用 tokio::spawn 函数创建一个异步任务（task）
    // run 函数接受 client 和 rx（接收端）作为参数
    // 并在 await 关键字处暂停执行，直到接收端接收到消息。
    tokio::spawn(async move { run(client, rx).await });

    // 返回缓冲区
    Buffer { tx }
}
```

##  Struct

###  Buffer

####  struct

```rust
/// 客户端的请求窗口
#[derive(Clone)]
pub struct Buffer {
    tx: Sender<Message>,
}
```

####  Implementation

```rust
impl Buffer{
    /// 获取key的值。请求会被缓存直到连接能够发送请求
        pub async fn get(&mut self, key: &str) -> Result<Option<Bytes>> {
        // 初始化一个get命令
        let get = Command::Get(key.into());

        // 初始化一次性的管道以接收连接的回应
        let (tx, rx) = oneshot::channel();

        // 发送请求
        self.tx.send((get, tx)).await?;

        // 等待回应，并返回结果
        match rx.await {
            Ok(res) => res,
            Err(err) => Err(err.into()),
      }
    }

    /// 设置key的值
    pub async fn set(&mut self, key: &str, value: Bytes) -> Result<()> {
        // 初始化一个set命令
        let set = Command::Set(key.into(), value);

        // 初始化一次性的管道以接收连接的回应
        let (tx, rx) = oneshot::channel();

        // 发送请求
        self.tx.send((set, tx)).await?;

        // 等待回应，并返回结果
        match rx.await {
            Ok(res) => res.map(|_| ()),
            Err(err) => Err(err.into()),
        }
    }           
}
```

###  Connection

####  struct

```rust
/// 发送和接收来自远端的帧
#[derive(Debug)]
pub struct Connection {
    // 带有写入缓冲区的tcp流
    stream: BufWriter<TcpStream>,

    // 帧读缓冲区
    buffer: BytesMut,
}
```

####  Implementation

```rust
impl Connection{
    /// 创建新的连接并返回socket，同时初始化缓冲区
    pub fn new(socket: TcpStream) -> Connection {
        Connection {
            stream: BufWriter::new(socket),
            buffer: BytesMut::with_capacity(4 * 1024),
        }
    }

    /// 从tcp流中读取一个帧
    pub async fn read_frame(&mut self) -> crate::Result<Option<Frame>> {
        loop {
            // 解析并获取帧
            if let Some(frame) = self.parse_frame()? {
                return Ok(Some(frame));
            }
            // 尝试从socket中读取更多的数据，如果读到的字节数为0，说明流被关闭
            if 0 == self.stream.read_buf(&mut self.buffer).await? {
                // 如果缓冲区为空说明远端正常关闭，否则为远端发送rst保温
                if self.buffer.is_empty() {
                    return Ok(None);
                } else {
                    return Err("connection reset by peer".into());
                }
            }
        }
    }

    /// 解析并获取frame。
    /// 如果数据不够则返回 `Ok(None)`
    /// 如果数据不合理则返回Err
    fn parse_frame(&mut self) -> crate::Result<Option<Frame>> {
        use frame::Error::Incomplete;

        // 创建cursor用于追踪buffer中的数据位置
        let mut buf = Cursor::new(&self.buffer[..]);

        // 检查缓冲区中是否有足够的数据
        match Frame::check(&mut buf) {
            Ok(_) => {
                // 由于Frame::check()开始时会定位到缓冲区的末尾，所以可以获取到缓冲区的数据长度
                let len = buf.position() as usize;

                // 重置cursor位置
                buf.set_position(0);

                // 解析frame
                let frame = Frame::parse(&mut buf)?;

                // 从buffer中删除所有数据（因为已经被解析了）
                self.buffer.advance(len);

                // 返回frame
                Ok(Some(frame))
            }
            // 数据不足以解析成单独的帧，返回None，说明需要继续等待
            Err(Incomplete) => Ok(None),
            // 返回错误
            Err(e) => Err(e.into()),
        }
    }

    /// 向TCP流中写数据
    pub async fn write_frame(&mut self, frame: &Frame) -> io::Result<()> {
        match frame {
            Frame::Array(val) => {
                // 对帧类型前缀进行编码。对于数组，它是“*”。
                self.stream.write_u8(b'*').await?;

                // 写入数组长度
                self.write_decimal(val.len() as u64).await?;

                // 遍历并写入数组中每个元素
                for entry in &**val {
                    self.write_value(entry).await?;
                }
            }
            // 直接写入整个帧
            _ => self.write_value(frame).await?,
        }

        // 保证缓冲区都被写入套接字
        self.stream.flush().await
    }

    /// 将帧文字写入流
    async fn write_value(&mut self, frame: &Frame) -> io::Result<()> {
            // 判断帧的类型，将不同的帧按照格式写入流
            match frame {
            Frame::Simple(val) => {
                self.stream.write_u8(b'+').await?;
                self.stream.write_all(val.as_bytes()).await?;
                self.stream.write_all(b"\r\n").await?;
            }
            Frame::Error(val) => {
                self.stream.write_u8(b'-').await?;
                self.stream.write_all(val.as_bytes()).await?;
                self.stream.write_all(b"\r\n").await?;
            }
            Frame::Integer(val) => {
                self.stream.write_u8(b':').await?;
                self.write_decimal(*val).await?;
            }
            Frame::Null => {
                self.stream.write_all(b"$-1\r\n").await?;
            }
            Frame::Bulk(val) => {
                let len = val.len();

                self.stream.write_u8(b'$').await?;
                self.write_decimal(len as u64).await?;
                self.stream.write_all(val).await?;
                self.stream.write_all(b"\r\n").await?;
            }
            // 暂时不支持解析数组类型
            Frame::Array(_val) => unreachable!(),
        }

        Ok(())
    }

    /// 将十进制帧写入流
    async fn write_decimal(&mut self, val: u64) -> io::Result<()> {
        use std::io::Write;

        // 将数据转换成string
        let mut buf = [0u8; 20];
        let mut buf = Cursor::new(&mut buf[..]);
        write!(&mut buf, "{}", val)?;
        // 将数据写入流
        let pos = buf.position() as usize;
        self.stream.write_all(&buf.get_ref()[..pos]).await?;
        self.stream.write_all(b"\r\n").await?;

        Ok(())
    }
}
```