Add TCP client loop for CLI

Signed-off-by: Chaitanya Munukutla <[email protected]>

Cargo.toml

@@ -3,7 +3,7 @@ members = ["server", "cli", "sdk"]
 resolver = "2"
 
 [workspace.package]
-version = "0.2.1"
+version = "0.2.2"
 license-file = "LICENSE.md"
 repository = "https://github.com/c16a/pouch"
 homepage = "https://github.com/c16a/pouch"

README.md

@@ -2,6 +2,14 @@
 
 A really tiny KV store.
 
+## Crates
+
+| Crate  | Version                                                                                                    |
+|--------|------------------------------------------------------------------------------------------------------------|
+| Server | [![pouch-server](https://img.shields.io/crates/v/pouch-server.svg)](https://crates.io/crates/pouch-server) |
+| CLI    | [![pouch-server](https://img.shields.io/crates/v/pouch-cli.svg)](https://crates.io/crates/pouch-cli)       |
+| SDK    | [![pouch-server](https://img.shields.io/crates/v/pouch-sdk.svg)](https://crates.io/crates/pouch-sdk)       |
+
 ## Building from source
 
 Both the `server` and the `cli` can be built from the repository root. The binaries can be found at `target/release`
@@ -10,6 +18,7 @@ or `target/debug` depending on the profile.
 ## Running tests
 
 All tests can be run from the repository root.
+
 ```shell
 cargo test
 ```

cli/Cargo.toml

@@ -7,4 +7,7 @@ keywords.workspace = true
 repository.workspace = true
 homepage.workspace = true
 license-file.workspace = true
-readme.workspace = true
+readme.workspace = true
+
+[dependencies]
+tokio = { version = "1.39.2", default-features = false, features = ["tokio-macros", "net", "rt-multi-thread", "macros", "io-util", "io-std"] }

cli/src/main.rs

@@ -1 +1,70 @@
-fn main() {}
+use std::env;
+use std::process::exit;
+use tokio::io;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::TcpStream;
+
+#[tokio::main]
+async fn main() {
+    let args: Vec<String> = env::args().collect();
+    if args.len() != 3 {
+        eprintln!("Usage: {} <server_address> <port>", args[0]);
+        exit(1);
+    }
+
+    let server_address = format!("{}:{}", args[1], args[2]);
+
+    let stream = TcpStream::connect(&server_address)
+        .await
+        .unwrap_or_else(|e| {
+            eprintln!("Failed to connect: {}", e);
+            exit(1);
+        });
+    println!("Connected to {}", server_address);
+
+    handle_interactive_loop(stream).await;
+}
+
+async fn handle_interactive_loop(mut stream: TcpStream) {
+    let (read_half, mut write_half) = stream.split();
+    let mut reader = BufReader::new(read_half).lines();
+    let stdin = io::stdin();
+    let mut stdin_reader = BufReader::new(stdin).lines();
+
+    let mut stdout = io::stdout();
+
+    loop {
+        stdout.write_all("pouch-cli> ".to_string().as_bytes()).await.unwrap();
+        stdout.flush().await.unwrap();
+
+        tokio::select! {
+            Ok(Some(command)) = stdin_reader.next_line() => {
+                if command.trim().is_empty() {
+                    continue;
+                }
+                if command.to_lowercase() == "quit" {
+                    println!("Exiting...");
+                    break;
+                }
+
+                // Send the command to the server
+                // Don't write a new line explicitly because the user would hit the ENTER key themselves.
+                write_half.write_all(command.as_bytes()).await.expect("Failed to write to server");
+
+                // Wait for the server response and print it back
+                if let Ok(Some(response)) = reader.next_line().await {
+                    stdout.write_all(format!("pouch-server> {}\n", response).as_bytes()).await.unwrap();
+                    stdout.flush().await.unwrap();
+                } else {
+                    stdout.write_all(b"No response from server.\n").await.unwrap();
+                    stdout.flush().await.unwrap();
+                }
+            }
+
+            Ok(Some(response)) = reader.next_line() => {
+                println!("server> {}", response);
+                io::stdout().flush().await.expect("Failed to initialise prompt");
+            }
+        }
+    }
+}

sdk/src/response.rs

@@ -1,5 +1,5 @@
-use std::collections::HashSet;
 use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub enum Response {

server/Cargo.toml

@@ -15,5 +15,5 @@ rand = "0.8.5"
 serde = { version = "1.0.204", default-features = false, features = ["derive", "std"] }
 serde_json = { version = "1.0.122", default-features = false, features = ["alloc"] }
 tokio = { version = "1.39.2", default-features = false, features = ["tokio-macros", "net", "rt-multi-thread", "sync", "macros", "io-util"] }
-pouch-sdk = { path = "../sdk", version = "0.2.1" }
+pouch-sdk = { path = "../sdk", version = "0.2.2" }
 

server/src/main.rs

@@ -1,6 +1,7 @@
 use std::env;
 use std::ops::DerefMut;
 use std::sync::Arc;
+use serde_json::json;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::RwLock;
@@ -97,7 +98,7 @@ async fn process(mut socket: TcpStream, db: Arc<RwLock<dyn Processor>>, wal: Arc
         let json_str = response.to_json().unwrap();
 
         // Write the response to the socket
-        if let Err(err) = socket.write_all(json_str.as_bytes()).await {
+        if let Err(err) = socket.write_all(format!("{}\n", json_str).as_bytes()).await {
             eprintln!("failed to write data to socket; err = {:?}", err);
             return;
         }
@@ -107,4 +108,3 @@ async fn process(mut socket: TcpStream, db: Arc<RwLock<dyn Processor>>, wal: Arc
         buf.resize(1024, 0); // Reset the buffer size
     }
 }
-

server/src/processor/db.rs

@@ -1,10 +1,10 @@
 use crate::processor::spec::Processor;
-use pouch_sdk::response::Response;
 use crate::structures::sorted_set::SortedSet;
 use crate::wal::WAL;
 use dashmap::mapref::one::{Ref, RefMut};
 use dashmap::DashMap;
 use pouch_sdk::command::Command;
+use pouch_sdk::response::Response;
 use std::collections::HashSet;
 use std::io;
 

server/src/processor/lists.rs

@@ -1,7 +1,7 @@
 use crate::processor::db::{DbValue, InMemoryDb};
+use dashmap::mapref::one::{Ref, RefMut};
 use pouch_sdk::response::Error::{IncompatibleDataType, UnknownKey};
 use pouch_sdk::response::Response;
-use dashmap::mapref::one::{Ref, RefMut};
 
 impl InMemoryDb {
     fn get_list_ref(&self, key: &String) -> Option<Ref<String, DbValue>> {

server/src/processor/sets.rs

@@ -1,7 +1,7 @@
 use crate::processor::db::{DbValue, InMemoryDb};
+use dashmap::mapref::one::Ref;
 use pouch_sdk::response::Error::{IncompatibleDataType, UnknownKey};
 use pouch_sdk::response::Response;
-use dashmap::mapref::one::Ref;
 use std::collections::HashSet;
 
 impl InMemoryDb {

server/src/processor/sorted_sets.rs

@@ -1,8 +1,8 @@
 use crate::processor::db::{DbValue, InMemoryDb};
-use pouch_sdk::response::Error::{IncompatibleDataType, UnknownKey};
-use pouch_sdk::response::Response;
 use crate::structures::sorted_set::{SortedSet, SortedSetAddReturnType};
 use dashmap::mapref::one::{Ref, RefMut};
+use pouch_sdk::response::Error::{IncompatibleDataType, UnknownKey};
+use pouch_sdk::response::Response;
 use std::collections::HashMap;
 
 impl InMemoryDb {

server/src/processor/spec.rs

@@ -1,6 +1,6 @@
-use pouch_sdk::response::Response;
 use crate::wal::WAL;
 use pouch_sdk::command::Command;
+use pouch_sdk::response::Response;
 
 pub(crate) trait Processor: Send + Sync {
     fn cmd(&self, cmd: Command, wal: Option<&mut WAL>) -> Response;

server/src/processor/strings.rs

@@ -1,6 +1,6 @@
 use crate::processor::db::{DbValue, InMemoryDb};
-use pouch_sdk::response::Error::{IncompatibleDataType, NotInteger, UnknownKey};
 use dashmap::mapref::one::Ref;
+use pouch_sdk::response::Error::{IncompatibleDataType, NotInteger, UnknownKey};
 use pouch_sdk::response::{Response, OK};
 
 impl InMemoryDb {