apply client/cli polish (#15)

Continuation of #11. Refines the client structure and implements GET.

`clap` is decoupled from the lib code. This is done to avoid any CLI
parsing concerns to leak into the lib. The main motivation for this is
to allow the reader to focus on Tokio concerns and not CLI parsing
concerns.

src/cmd/get.rs

@@ -1,6 +1,6 @@
 use crate::{Connection, Db, Frame, Parse, ParseError};
 
-use std::io;
+use bytes::Bytes;
 use tracing::{debug, instrument};
 
 #[derive(Debug)]
@@ -9,10 +9,14 @@ pub struct Get {
 }
 
 impl Get {
+    /// Create a new `Get` command which fetches `key`.
+    pub(crate) fn new(key: impl ToString) -> 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
-    #[instrument]
     pub(crate) fn parse_frames(parse: &mut Parse) -> Result<Get, ParseError> {
         let key = parse.next_string()?;
 
@@ -26,14 +30,24 @@ impl Get {
         Ok(Get { key })
     }
 
-    #[instrument]
-    pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> io::Result<()> {
+    #[instrument(skip(self, db, dst))]
+    pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> crate::Result<()> {
         let response = if let Some(value) = db.get(&self.key) {
             Frame::Bulk(value)
         } else {
             Frame::Null
         };
+
         debug!(?response);
-        dst.write_frame(&response).await
+
+        dst.write_frame(&response).await?;
+        Ok(())
+    }
+
+    pub(crate) fn into_frame(self) -> Frame {
+        let mut frame = Frame::array();
+        frame.push_bulk(Bytes::from("get".as_bytes()));
+        frame.push_bulk(Bytes::from(self.key.into_bytes()));
+        frame
     }
 }

src/cmd/mod.rs

@@ -10,13 +10,8 @@ pub use set::Set;
 mod subscribe;
 pub use subscribe::{Subscribe, Unsubscribe};
 
-pub(crate) mod utils;
-
 use crate::{Connection, Db, Frame, Parse, ParseError, Shutdown};
 
-use std::io;
-use tracing::instrument;
-
 #[derive(Debug)]
 pub(crate) enum Command {
     Get(Get),
@@ -27,7 +22,6 @@ pub(crate) enum Command {
 }
 
 impl Command {
-    #[instrument]
     pub(crate) fn from_frame(frame: Frame) -> Result<Command, ParseError> {
         let mut parse = Parse::new(frame)?;
 
@@ -46,20 +40,12 @@ impl Command {
         Ok(command)
     }
 
-    pub(crate) fn into_frame(self) -> Result<Frame, ParseError> {
-        let frame = match self {
-            Command::Set(set) => set.into_frame(),
-            _ => unimplemented!(),
-        };
-        Ok(frame)
-    }
-
     pub(crate) async fn apply(
         self,
         db: &Db,
         dst: &mut Connection,
         shutdown: &mut Shutdown,
-    ) -> io::Result<()> {
+    ) -> crate::Result<()> {
         use Command::*;
 
         match self {

src/cmd/publish.rs

@@ -1,7 +1,6 @@
 use crate::{Connection, Db, Frame, Parse, ParseError};
 
 use bytes::Bytes;
-use std::io;
 
 #[derive(Debug)]
 pub struct Publish {
@@ -17,11 +16,12 @@ impl Publish {
         Ok(Publish { channel, message })
     }
 
-    pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> io::Result<()> {
+    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);
 
         let response = Frame::Integer(num_subscribers as u64);
-        dst.write_frame(&response).await
+        dst.write_frame(&response).await?;
+        Ok(())
     }
 }

src/cmd/set.rs

@@ -1,26 +1,19 @@
-use crate::cmd::{
-    utils::{bytes_from_str, duration_from_ms_str},
-    Parse, ParseError,
-};
 use crate::{Connection, Db, Frame};
-use clap::Clap;
+use crate::cmd::{Parse, ParseError};
 
 use bytes::Bytes;
-use std::io;
 use std::time::Duration;
 use tracing::{debug, instrument};
 
-#[derive(Clap, Debug)]
+#[derive(Debug)]
 pub struct Set {
     /// the lookup key
     pub(crate) key: String,
 
     /// the value to be stored
-    #[clap(parse(from_str = bytes_from_str))]
     pub(crate) value: Bytes,
 
-    /// duration in milliseconds
-    #[clap(parse(try_from_str = duration_from_ms_str))]
+    /// When to expire the key
     pub(crate) expire: Option<Duration>,
 }
 
@@ -52,14 +45,15 @@ impl Set {
         Ok(Set { key, value, expire })
     }
 
-    #[instrument]
-    pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> io::Result<()> {
+    #[instrument(skip(db))]
+    pub(crate) async fn apply(self, db: &Db, dst: &mut Connection) -> crate::Result<()> {
         // Set the value
         db.set(self.key, self.value, self.expire);
 
         let response = Frame::Simple("OK".to_string());
         debug!(?response);
-        dst.write_frame(&response).await
+        dst.write_frame(&response).await?;
+        Ok(())
     }
 
     pub(crate) fn into_frame(self) -> Frame {

src/cmd/subscribe.rs

@@ -1,8 +1,7 @@
-use crate::cmd::{Parse, ParseError};
 use crate::{Command, Connection, Db, Frame, Shutdown};
+use crate::cmd::{Parse, ParseError};
 
 use bytes::Bytes;
-use std::io;
 use tokio::select;
 use tokio::stream::{StreamExt, StreamMap};
 
@@ -48,7 +47,7 @@ impl Subscribe {
         db: &Db,
         dst: &mut Connection,
         shutdown: &mut Shutdown,
-    ) -> io::Result<()> {
+    ) -> 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.

src/cmd/utils.rs

@@ -1,11 +0,0 @@
-use bytes::Bytes;
-use std::time::Duration;
-
-pub(crate) fn duration_from_ms_str(src: &str) -> Result<Duration, std::num::ParseIntError> {
-    let millis = src.parse::<u64>()?;
-    Ok(Duration::from_millis(millis))
-}
-
-pub(crate) fn bytes_from_str(src: &str) -> Bytes {
-    Bytes::from(src.to_string())
-}