feat: build client

Author: odysa

src/client.rs

@@ -0,0 +1,83 @@
+use std::{
+    io::{BufReader, BufWriter, Write},
+    net::{SocketAddr, TcpStream},
+};
+
+use serde_json::{de::IoRead, StreamDeserializer};
+
+use crate::{
+    error::{Error, ErrorKind, Result},
+    net::{Request, Response},
+};
+
+pub struct Client<'a> {
+    writer: BufWriter<TcpStream>,
+    reader: StreamDeserializer<'a, IoRead<BufReader<TcpStream>>, Response>,
+}
+
+impl<'a> Client<'a> {
+    pub fn connect(addr: SocketAddr) -> Result<Client<'a>> {
+        let stream = TcpStream::connect(addr)?;
+        let writer = BufWriter::new(stream.try_clone()?);
+        let reader = BufReader::new(stream);
+        let reader = serde_json::Deserializer::from_reader(reader).into_iter::<Response>();
+        Ok(Client { reader, writer })
+    }
+
+    pub fn get(&mut self, key: String) -> Result<Option<String>> {
+        let request = Request::Get {
+            key: key.to_owned(),
+        };
+
+        self.send_request(&request)?;
+
+        if let Some(response) = self.reader.next() {
+            if let Response::Get(Ok(result)) = response? {
+                return Ok(result);
+            }
+        }
+
+        Err(Error::from(ErrorKind::KeyNotFound(format!(
+            "cannot get value of key:{}",
+            key
+        ))))
+    }
+
+    pub fn set(&mut self, key: String, value: String) -> Result<()> {
+        let request = Request::Set { key, value };
+        self.send_request(&request)?;
+
+        if let Some(response) = self.reader.next() {
+            if let Response::Set(Ok(())) = response? {
+                return Ok(());
+            }
+        }
+        Err(Error::from(ErrorKind::Error(
+            "cannot get response from server".to_string(),
+        )))
+    }
+
+    pub fn remove(&mut self, key: String) -> Result<()> {
+        let request = Request::Remove { key };
+        self.send_request(&request)?;
+
+        if let Some(response) = self.reader.next() {
+            if let Response::Remove(Ok(())) = response? {
+                return Ok(());
+            }
+        }
+
+        Err(Error::from(ErrorKind::Error(
+            "cannot get response from server".to_string(),
+        )))
+    }
+
+    fn send_request(&mut self, request: &Request) -> Result<()> {
+        let buf = serde_json::to_vec(request)?;
+
+        self.writer.write_all(&buf[..])?;
+        self.writer.flush()?;
+
+        Ok(())
+    }
+}

src/common.rs

@@ -57,20 +57,3 @@ impl OffSet {
         self.file_no
     }
 }
-#[derive(Debug, Serialize)]
-pub enum GetResponse {
-    Ok(Option<String>),
-    Err(String),
-}
-
-#[derive(Debug, Serialize)]
-pub enum SetResponse {
-    Ok(()),
-    Err(String),
-}
-
-#[derive(Debug, Serialize)]
-pub enum RemoveResponse {
-    Ok(()),
-    Err(String),
-}

src/lib.rs

@@ -1,7 +1,9 @@
+mod client;
 pub mod common;
 mod database;
 pub mod error;
 pub mod kvs_store;
+mod net;
 mod protocol;
 mod reader;
 pub mod server;

src/net.rs

@@ -0,0 +1,30 @@
+use serde::{Deserialize, Serialize};
+use std::fs::OpenOptions;
+
+#[derive(Serialize, Deserialize, Debug)]
+pub enum Request {
+    Get { key: String },
+    Set { key: String, value: String },
+    Remove { key: String },
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub enum Response {
+    Get(Result<Option<String>, String>),
+    Set(Result<(), String>),
+    Remove(Result<(), String>),
+}
+
+impl Response {
+    pub fn set(result: Result<(), String>) -> Self {
+        Response::Set(result)
+    }
+
+    pub fn get(result: Result<Option<String>, String>) -> Self {
+        Response::Get(result)
+    }
+
+    pub fn remove(result: Result<(), String>) -> Self {
+        Response::Remove(result)
+    }
+}

src/protocol/connection.rs

@@ -1,12 +1,12 @@
-use std::io::{BufReader, Cursor};
+use std::io::Cursor;
 
-use bytes::{buf, Buf, BytesMut};
+use bytes::{Buf, BytesMut};
 use tokio::{
-    io::{AsyncReadExt, AsyncWriteExt, BufWriter},
+    io::{AsyncWriteExt, BufWriter},
     net::TcpStream,
 };
 
-use crate::error::{Error, Result};
+use crate::error::Result;
 
 use super::protocol::Frame;
 

src/server.rs

@@ -1,5 +1,6 @@
-use crate::common::{Command, GetResponse, KvsEngine, RemoveResponse, SetResponse};
+use crate::common::KvsEngine;
 use crate::error::Result;
+use crate::net::{Request, Response};
 use serde_json::Deserializer;
 use slog::{debug, error, info, Logger};
 use std::io::{BufReader, BufWriter, Write};
@@ -20,55 +21,51 @@ impl<T: KvsEngine> Server<T> {
         }
         Ok(())
     }
-
     fn handle_client(&mut self, stream: TcpStream, logger: &Logger) -> Result<()> {
         let peer_addr = stream.peer_addr()?;
         let reader = BufReader::new(&stream);
         let mut writer = BufWriter::new(&stream);
-        let requests = Deserializer::from_reader(reader).into_iter::<Command>();
+        let requests = Deserializer::from_reader(reader).into_iter::<Request>();
+
+        for request in requests {
+            if let Ok(request) = request {
+                info!(logger,"request:"; "request" => format!("{:?}", request));
+
+                let response = match request {
+                    Request::Get { key } => match self.engine.get(key) {
+                        Ok(v) => Response::Get(Ok(v)),
+                        Err(e) => Response::Get(Err(e.to_string())),
+                    },
+                    Request::Remove { key } => match self.engine.remove(key) {
+                        Ok(_) => Response::Remove(Ok(())),
+                        Err(e) => Response::Remove(Err(e.as_string())),
+                    },
+                    Request::Set { key, value } => match self.engine.set(key, value) {
+                        Ok(()) => Response::Set(Ok(())),
+                        Err(e) => Response::Set(Err(e.to_string())),
+                    },
+                };
+
+                self.send_response(&mut writer, &response)?;
 
-        macro_rules! respond {
-            ($response: expr) => {{
-                let response = $response;
-                serde_json::to_writer(&mut writer, &response)?;
-                writer.flush()?;
                 debug!(
                     logger,
                     "Response sent to";
                     "addr" => format!("{}",peer_addr),
                     "response" => format!("{:?}",response)
                 );
-            };};
-        }
-
-        for request in requests {
-            if let Ok(request) = request {
-                info!(logger,"request:"; "request" => format!("{:?}", request));
-                match request {
-                    Command::Get { key } => {
-                        respond!(match self.engine.get(key) {
-                            Ok(v) => GetResponse::Ok(v),
-                            Err(e) => GetResponse::Err(e.as_string()),
-                        })
-                    }
-                    Command::Remove { key } => {
-                        respond!(match self.engine.remove(key) {
-                            Ok(_) => RemoveResponse::Ok(()),
-                            Err(e) => RemoveResponse::Err(e.as_string()),
-                        })
-                    }
-                    Command::Set { key, value } => {
-                        respond!(match self.engine.set(key, value) {
-                            Ok(()) => SetResponse::Ok(()),
-                            Err(e) => SetResponse::Err(e.as_string()),
-                        })
-                    }
-                }
-                writer.flush()?;
+                // write response
             } else {
                 error!(logger, "can not parse the request");
             }
         }
         Ok(())
     }
+
+    fn send_response(&self, writer: &mut BufWriter<&TcpStream>, response: &Response) -> Result<()> {
+        let buf = serde_json::to_vec(response)?;
+        writer.write_all(&buf[..])?;
+        writer.flush()?;
+        Ok(())
+    }
 }