mostly docs, some code tweaks as well (#31)

Db background tasks never shutdown o_O

Author: carllerche

README.md

@@ -112,6 +112,11 @@ the server to update the active subscriptions.
 [broadcast]: https://docs.rs/tokio/*/tokio/sync/broadcast/index.html
 [`StreamMap`]: https://docs.rs/tokio/*/tokio/stream/struct.StreamMap.html
 
+### Using a `std::sync::Mutex` in an async application
+
+The server uses a `std::sync::Mutex` and **not** a Tokio mutex to synchronize
+access to shared state. See [`db.rs`](src/db.rs) for more details.
+
 ## Contributing
 
 Contributions to `mini-redis` are welcome. Keep in mind, the goal of the project

src/client.rs

@@ -47,12 +47,7 @@ impl Client {
     /// Set the value of a key to `value`.
     #[instrument(skip(self))]
     pub async fn set(&mut self, key: &str, value: Bytes) -> crate::Result<()> {
-        self.set_cmd(Set {
-            key: key.to_string(),
-            value: value,
-            expire: None,
-        })
-        .await
+        self.set_cmd(Set::new(key, value, None)).await
     }
 
     /// publish `message` on the `channel`
@@ -88,12 +83,7 @@ impl Client {
         value: Bytes,
         expiration: Duration,
     ) -> crate::Result<()> {
-        self.set_cmd(Set {
-            key: key.to_string(),
-            value: value.into(),
-            expire: Some(expiration),
-        })
-        .await
+        self.set_cmd(Set::new(key, value, Some(expiration))).await
     }
 
     async fn set_cmd(&mut self, cmd: Set) -> crate::Result<()> {

src/cmd/get.rs

@@ -1,10 +1,16 @@
-use crate::{Connection, Db, Frame, Parse, ParseError};
+use crate::{Connection, Db, Frame, Parse};
 
 use bytes::Bytes;
 use tracing::{debug, instrument};
 
+/// Get the value of key.
+///
+/// If the key does not exist the special value nil is returned. An error is
+/// returned if the value stored at key is not a string, because GET only
+/// handles string values.
 #[derive(Debug)]
 pub struct Get {
+    /// Name of the key to get
     key: String,
 }
 
@@ -14,36 +20,63 @@ impl Get {
         Get { key: key.to_string() }
     }
 
-    // instrumenting functions will log all of the arguments passed to the function
-    // with their debug implementations
-    // see https://docs.rs/tracing/0.1.13/tracing/attr.instrument.html
-    pub(crate) fn parse_frames(parse: &mut Parse) -> Result<Get, ParseError> {
+    /// Parse a `Get` instance from received data.
+    ///
+    /// The `Parse` argument provides a cursor like API to read fields from a
+    /// received `Frame`. At this point, the data has already been received from
+    /// the socket.
+    ///
+    /// The `GET` string has already been consumed.
+    ///
+    /// # Returns
+    ///
+    /// Returns the `Get` value on success. If the frame is malformed, `Err` is
+    /// returned.
+    ///
+    /// # Format
+    ///
+    /// Expects an array frame containing two entries.
+    ///
+    /// ```text
+    /// GET key
+    /// ```
+    pub(crate) fn parse_frames(parse: &mut Parse) -> crate::Result<Get> {
+        // The `GET` string has already been consumed. The next value is the
+        // name of the key to get. If the next value is not a string or the
+        // input is fully consumed, then an error is returned.
         let key = parse.next_string()?;
 
-        // adding this debug event allows us to see what key is parsed
-        // the ? sigil tells `tracing` to use the `Debug` implementation
-        // get parse events can be filtered by running
-        // RUST_LOG=mini_redis::cmd::get[parse_frames]=debug cargo run --bin server
-        // see https://docs.rs/tracing/0.1.13/tracing/#recording-fields
-        debug!(?key);
-
         Ok(Get { key })
     }
 
+    /// Apply the `Get` command to the specified `Db` instance.
+    ///
+    /// The response is written to `dst`. This is called by the server in order
+    /// to execute a received command.
     #[instrument(skip(self, db, dst))]
     pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> crate::Result<()> {
+        // Get the value from the shared database state
         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);
 
+        // Write the response back to the client
         dst.write_frame(&response).await?;
+
         Ok(())
     }
 
+    /// Converts the command into an equivalent `Frame`.
+    ///
+    /// This is called by the client when encoding a `Get` command to send to
+    /// the server.
     pub(crate) fn into_frame(self) -> Frame {
         let mut frame = Frame::array();
         frame.push_bulk(Bytes::from("get".as_bytes()));

src/cmd/mod.rs

@@ -15,6 +15,9 @@ pub use unknown::Unknown;
 
 use crate::{Connection, Db, Frame, Parse, ParseError, Shutdown};
 
+/// Enumeration of supported Redis commands.
+///
+/// Methods called on `Command` are delegated to the command implementation.
 #[derive(Debug)]
 pub(crate) enum Command {
     Get(Get),
@@ -26,27 +29,59 @@ pub(crate) enum Command {
 }
 
 impl Command {
+    /// Parse a command from a received frame.
+    ///
+    /// The `Frame` must represent a Redis command supported by `mini-redis` and
+    /// be the array variant.
+    ///
+    /// # Returns
+    ///
+    /// On success, the command value is returned, otherwise, `Err` is returned.
     pub(crate) fn from_frame(frame: Frame) -> crate::Result<Command> {
+        // The frame  value is decorated with `Parse`. `Parse` provides a
+        // "cursor" like API which makes parsing the command easier.
+        //
+        // The frame value must be an array variant. Any other frame variants
+        // result in an error being returned.
         let mut parse = Parse::new(frame)?;
 
+        // All redis commands begin with the command name as a string. The name
+        // is read and converted to lower casae in order to do case sensitive
+        // matching.
         let command_name = parse.next_string()?.to_lowercase();
 
+        // Match the command name, delegating the rest of the parsing to the
+        // specific command.
         let command = match &command_name[..] {
             "get" => Command::Get(Get::parse_frames(&mut parse)?),
             "publish" => Command::Publish(Publish::parse_frames(&mut parse)?),
             "set" => Command::Set(Set::parse_frames(&mut parse)?),
             "subscribe" => Command::Subscribe(Subscribe::parse_frames(&mut parse)?),
             "unsubscribe" => Command::Unsubscribe(Unsubscribe::parse_frames(&mut parse)?),
             _ => {
-                parse.next_string()?;
-                Command::Unknown(Unknown::new(command_name))
+                // The command is not recognized and an Unknown command is
+                // returned.
+                //
+                // `return` is called here to skip the `finish()` call below. As
+                // the command is not recognized, there is most likely
+                // unconsumed fields remaining in the `Parse` instance.
+                return Ok(Command::Unknown(Unknown::new(command_name)));
             },
         };
 
+        // Check if there is any remaining unconsumed fields in the `Parse`
+        // value. If fields remain, this indicates an unexpected frame format
+        // and an error is returned.
         parse.finish()?;
+
+        // The command has been successfully parsed
         Ok(command)
     }
 
+    /// Apply the command to the specified `Db` instance.
+    ///
+    /// The response is written to `dst`. This is called by the server in order
+    /// to execute a received command.
     pub(crate) async fn apply(
         self,
         db: &Db,
@@ -63,10 +98,11 @@ impl Command {
             Unknown(cmd) => cmd.apply(dst).await,
             // `Unsubscribe` cannot be applied. It may only be received from the
             // context of a `Subscribe` command.
-            Unsubscribe(_) => unimplemented!(),
+            Unsubscribe(_) => Err("`Unsubscribe` is unsupported in this context".into()),
         }
     }
 
+    /// Returns the command name
     pub(crate) fn get_name(&self) -> &str {
         match self {
             Command::Get(_) => "get",

src/cmd/publish.rs

@@ -1,4 +1,4 @@
-use crate::{Connection, Db, Frame, Parse, ParseError};
+use crate::{Connection, Db, Frame, Parse};
 
 use bytes::Bytes;
 
@@ -9,13 +9,17 @@ pub struct Publish {
 }
 
 impl Publish {
-    pub(crate) fn parse_frames(parse: &mut Parse) -> Result<Publish, ParseError> {
+    pub(crate) fn parse_frames(parse: &mut Parse) -> crate::Result<Publish> {
         let channel = parse.next_string()?;
         let message = parse.next_bytes()?;
 
         Ok(Publish { channel, message })
     }
 
+    /// Apply the `Publish` command to the specified `Db` instance.
+    ///
+    /// The response is written to `dst`. This is called by the server in order
+    /// to execute a received command.
     pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> crate::Result<()> {
         // Set the value
         let num_subscribers = db.publish(&self.channel, self.message);

src/cmd/set.rs

@@ -5,57 +5,127 @@ use bytes::Bytes;
 use std::time::Duration;
 use tracing::{debug, instrument};
 
+/// Set `key` to hold the string `value`.
+///
+/// If `key` already holds a value, it is overwritten, regardless of its type.
+/// Any previous time to live associated with the key is discarded on successful
+/// SET operation.
+///
+/// # Options
+///
+/// Currently, the following options are supported:
+///
+/// * EX `seconds` -- Set the specified expire time, in seconds.
+/// * PX `milliseconds` -- Set the specified expire time, in milliseconds.
 #[derive(Debug)]
 pub struct Set {
     /// the lookup key
-    pub(crate) key: String,
+    key: String,
 
     /// the value to be stored
-    pub(crate) value: Bytes,
+    value: Bytes,
 
     /// When to expire the key
-    pub(crate) expire: Option<Duration>,
+    expire: Option<Duration>,
 }
 
 impl Set {
-    #[instrument]
-    pub(crate) fn parse_frames(parse: &mut Parse) -> Result<Set, ParseError> {
+    /// Create a new `Set` command which sets `key` to `value`.
+    ///
+    /// If `expire` is `Some`, the value should expire after the specified
+    /// duration.
+    pub(crate) fn new(key: impl ToString, value: Bytes, expire: Option<Duration>) -> Set {
+        Set {
+            key: key.to_string(),
+            value,
+            expire,
+        }
+    }
+
+    /// Parse a `Set` instance from received data.
+    ///
+    /// The `Parse` argument provides a cursor like API to read fields from a
+    /// received `Frame`. At this point, the data has already been received from
+    /// the socket.
+    ///
+    /// The `SET` string has already been consumed.
+    ///
+    /// # Returns
+    ///
+    /// Returns the `Set` value on success. If the frame is malformed, `Err` is
+    /// returned.
+    ///
+    /// # Format
+    ///
+    /// Expects an array frame containing at least 3 entries.
+    ///
+    /// ```text
+    /// SET key value [EX seconds|PX milliseconds]
+    /// ```
+    pub(crate) fn parse_frames(parse: &mut Parse) -> crate::Result<Set> {
         use ParseError::EndOfStream;
 
+        // Read the key to set. This is a required field
         let key = parse.next_string()?;
+
+        // Read the value to set. This is a required field.
         let value = parse.next_bytes()?;
+
+        // The expiration is optional. If nothing else follows, then it is
+        // `None`.
         let mut expire = None;
 
+        // Attempt to parse another string.
         match parse.next_string() {
             Ok(s) if s == "EX" => {
+                // An expiration is specified in seconds. The next value is an
+                // integer.
                 let secs = parse.next_int()?;
                 expire = Some(Duration::from_secs(secs));
             }
             Ok(s) if s == "PX" => {
+                // An expiration is specified in milliseconds. The next value is
+                // an integer.
                 let ms = parse.next_int()?;
                 expire = Some(Duration::from_millis(ms));
             }
-            Ok(_) => unimplemented!(),
+            // Currently, mini-redis does not support any of the other SET
+            // options. An error here results in the connection being
+            // terminated. Other connections will continue to operate normally.
+            Ok(_) => return Err("currently `SET` only supports the expiration option".into()),
+            // The `EndOfStream` error indicates there is no further data to
+            // parse. In this case, it is a normal run time situation and
+            // indicates there are no specified `SET` options.
             Err(EndOfStream) => {}
-            Err(err) => return Err(err),
+            // All other errors are bubbled up, resulting in the connection
+            // being terminated.
+            Err(err) => return Err(err.into()),
         }
 
-        debug!(?key, ?value, ?expire);
-
         Ok(Set { key, value, expire })
     }
 
-    #[instrument(skip(db))]
+    /// Apply the `Get` command to the specified `Db` instace.
+    ///
+    /// The response is written to `dst`. This is called by the server in order
+    /// to execute a received command.
+    #[instrument(skip(self, db, dst))]
     pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> crate::Result<()> {
-        // Set the value
+        // Set the value in the shared database state.
         db.set(self.key, self.value, self.expire);
 
+        // Create a success response and write it to `dst`.
         let response = Frame::Simple("OK".to_string());
         debug!(?response);
         dst.write_frame(&response).await?;
+
         Ok(())
     }
 
+    /// Converts the command into an equivalent `Frame`.
+    ///
+    /// This is called by the client when encoding a `Set` command to send to
+    /// the server.
     pub(crate) fn into_frame(self) -> Frame {
         let mut frame = Frame::array();
         frame.push_bulk(Bytes::from("set".as_bytes()));