Add the `!pipe` command to the shell

Cargo.lock

@@ -394,9 +394,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
 
 [[package]]
 name = "libc"
-version = "0.2.106"
+version = "0.2.107"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a60553f9a9e039a333b4e9b20573b9e9b9c0bb3a11e201ccc48ef4283456d673"
+checksum = "fbe5e23404da5b4f555ef85ebed98fb4083e55a00c317800bc2a50ede9f3d219"
 
 [[package]]
 name = "libsky"
@@ -810,9 +810,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.68"
+version = "1.0.69"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0f690853975602e1bfe1ccbf50504d67174e3bcf340f23b5ea9992e0587a52d8"
+checksum = "e466864e431129c7e0d3476b92f20458e5879919a0596c6472738d9fa2d342f8"
 dependencies = [
  "itoa",
  "ryu",
@@ -859,7 +859,7 @@ dependencies = [
  "rand",
  "serde",
  "serde_json",
- "skytable 0.6.1 (git+https://github.com/skytable/client-rust?branch=next)",
+ "skytable 0.6.2-alpha.2 (git+https://github.com/skytable/client-rust?branch=next)",
 ]
 
 [[package]]
@@ -870,7 +870,7 @@ dependencies = [
  "clap",
  "env_logger",
  "log",
- "skytable 0.6.1 (git+https://github.com/skytable/client-rust.git)",
+ "skytable 0.6.2-alpha.2 (git+https://github.com/skytable/client-rust.git)",
 ]
 
 [[package]]
@@ -907,7 +907,7 @@ dependencies = [
  "regex",
  "serde",
  "sky_macros",
- "skytable 0.6.1 (git+https://github.com/skytable/client-rust?branch=next)",
+ "skytable 0.6.2-alpha.2 (git+https://github.com/skytable/client-rust?branch=next)",
  "tokio",
  "tokio-openssl",
  "toml",
@@ -922,14 +922,14 @@ dependencies = [
  "crossterm",
  "libsky",
  "rustyline",
- "skytable 0.6.1 (git+https://github.com/skytable/client-rust?branch=next)",
+ "skytable 0.6.2-alpha.2 (git+https://github.com/skytable/client-rust?branch=next)",
  "tokio",
 ]
 
 [[package]]
 name = "skytable"
-version = "0.6.1"
+version = "0.6.2-alpha.2"
-source = "git+https://github.com/skytable/client-rust?branch=next#9a34fd825a22d172e61786783c7c4606b6e38e8a"
+source = "git+https://github.com/skytable/client-rust?branch=next#43ab1e47903a3772657351bc5dea64cc4258f491"
 dependencies = [
  "bytes",
  "openssl",
@@ -939,8 +939,8 @@ dependencies = [
 
 [[package]]
 name = "skytable"
-version = "0.6.1"
+version = "0.6.2-alpha.2"
-source = "git+https://github.com/skytable/client-rust.git#9a34fd825a22d172e61786783c7c4606b6e38e8a"
+source = "git+https://github.com/skytable/client-rust.git#43ab1e47903a3772657351bc5dea64cc4258f491"
 
 [[package]]
 name = "smallvec"
@@ -965,7 +965,7 @@ dependencies = [
  "log",
  "num_cpus",
  "rand",
- "skytable 0.6.1 (git+https://github.com/skytable/client-rust?branch=next)",
+ "skytable 0.6.2-alpha.2 (git+https://github.com/skytable/client-rust?branch=next)",
  "sysinfo",
 ]
 

cli/src/argparse.rs

@@ -25,6 +25,7 @@
 */
 
 use crate::runner::Runner;
+use crate::tokenizer;
 use clap::load_yaml;
 use clap::App;
 use crossterm::terminal::{Clear, ClearType};
@@ -34,6 +35,8 @@ use libsky::VERSION;
 use readline::config::Configurer;
 use readline::{error::ReadlineError, Editor};
 use rustyline as readline;
+use skytable::Pipeline;
+use skytable::Query;
 use std::io::stdout;
 use std::process;
 const ADDR: &str = "127.0.0.1";
@@ -69,6 +72,7 @@ the server. These enable you to do convenient things like:
 - "clear": clears the terminal screen
 
 Apart from these, you can use the following shell commands:
+- "!pipe": Lets you create a pipeline. Terminate with a semicolon (`;`)
 - "!help": Brings up this help menu
 - "?<command name>": Describes what the built-in shell command is for
 
@@ -132,14 +136,25 @@ pub async fn start_repl() {
     }
     loop {
         match editor.readline(SKYSH_PROMPT) {
-            Ok(mut line) => match line.to_lowercase().as_str() {
+            Ok(mut line) => {
+                macro_rules! tokenize {
+                    ($inp:expr) => {
+                        match tokenizer::get_query($inp) {
+                            Ok(q) => q,
+                            Err(e) => {
+                                eskysh!(e);
+                                continue;
+                            }
+                        }
+                    };
+                    () => {
+                        tokenize!(line.as_bytes())
+                    };
+                }
+                match line.to_lowercase().as_str() {
                     "exit" => break,
                     "clear" => {
-                    let mut stdout = stdout();
+                        clear_screen();
-                    execute!(stdout, Clear(ClearType::All)).expect("Failed to clear screen");
-                    execute!(stdout, cursor::MoveTo(0, 0))
-                        .expect("Failed to move cursor to origin");
-                    drop(stdout); // aggressively drop stdout
                         continue;
                     }
                     "help" => {
@@ -153,23 +168,38 @@ pub async fn start_repl() {
                         match line.as_bytes()[0] {
                             b'#' => continue,
                             b'!' => {
-                            // handle a shell command
                                 match &line.as_bytes()[1..] {
                                     b"" => eskysh!("Bad shell command"),
                                     b"help" => println!("{}", HELP_TEXT),
+                                    b"pipe" => {
+                                        // so we need to handle a pipeline
+                                        let mut pipeline = Pipeline::new();
+                                        line = readln!(editor);
+                                        loop {
+                                            if !line.is_empty() {
+                                                if *(line.as_bytes().last().unwrap()) == b';' {
+                                                    break;
+                                                } else {
+                                                    let q: Query = tokenize!();
+                                                    pipeline.push(q);
+                                                }
+                                            }
+                                            line = readln!(editor);
+                                        }
+                                        if line.len() > 1 {
+                                            line.drain(line.len() - 1..);
+                                            let q: Query = tokenize!();
+                                            pipeline.push(q);
+                                        }
+                                        runner.run_pipeline(pipeline).await;
+                                    }
                                     _ => eskysh!("Unknown shell command"),
                                 }
                                 continue;
                             }
                             b'?' => {
                                 // handle explanation for a shell command
-                            match &line.as_bytes()[1..] {
+                                print_help(&line);
-                                b"" => eskysh!("Bad shell command"),
-                                b"help" => println!("`!help` shows the help menu"),
-                                b"exit" => println!("`exit` ends the shell session"),
-                                b"clear" => println!("`clear` clears the terminal screen"),
-                                _ => eskysh!("Unknown shell command"),
-                            }
                                 continue;
                             }
                             _ => {}
@@ -182,7 +212,8 @@ pub async fn start_repl() {
                         }
                         runner.run_query(&line).await
                     }
-            },
+                }
+            }
             Err(ReadlineError::Interrupted) => break,
             Err(err) => fatal!("ERROR: Failed to read line with error: {}", err),
         }
@@ -194,3 +225,21 @@ pub async fn start_repl() {
         })
         .unwrap();
 }
+
+fn print_help(line: &str) {
+    match &line.as_bytes()[1..] {
+        b"" => eskysh!("Bad shell command"),
+        b"help" => println!("`!help` shows the help menu"),
+        b"exit" => println!("`exit` ends the shell session"),
+        b"clear" => println!("`clear` clears the terminal screen"),
+        b"pipe" | b"!pipe" => println!("`!pipe` lets you run pipelines using the shell"),
+        _ => eskysh!("Unknown shell command"),
+    }
+}
+
+fn clear_screen() {
+    let mut stdout = stdout();
+    execute!(stdout, Clear(ClearType::All)).expect("Failed to clear screen");
+    execute!(stdout, cursor::MoveTo(0, 0)).expect("Failed to move cursor to origin");
+    drop(stdout); // aggressively drop stdout
+}

cli/src/runner.rs

@@ -29,6 +29,7 @@ use crossterm::style::{Color, Print, ResetColor, SetForegroundColor};
 use skytable::error::Error;
 use skytable::types::Array;
 use skytable::types::FlatElement;
+use skytable::Pipeline;
 use skytable::Query;
 use skytable::{aio, Element, RespCode};
 
@@ -48,6 +49,24 @@ impl Runner {
         let con = aio::TlsConnection::new(host, port, cert).await?;
         Ok(Self::Secure(con))
     }
+    pub async fn run_pipeline(&mut self, pipeline: Pipeline) {
+        let ret = match self {
+            Self::Insecure(con) => con.run_pipeline(pipeline).await,
+            Self::Secure(con) => con.run_pipeline(pipeline).await,
+        };
+        let retok = match ret {
+            Ok(r) => r,
+            Err(e) => fatal!("An I/O error occurred while querying: {}", e),
+        };
+        for (idx, resp) in retok
+            .into_iter()
+            .enumerate()
+            .map(|(idx, resp)| (idx + 1, resp))
+        {
+            println!("[Response {}]", idx);
+            print_element(resp);
+        }
+    }
     pub async fn run_query(&mut self, unescaped: &str) {
         let query: Query = match tokenizer::get_query(unescaped.as_bytes()) {
             Ok(q) => q,
@@ -61,7 +80,14 @@ impl Runner {
             Self::Secure(con) => con.run_simple_query(&query).await,
         };
         match ret {
-            Ok(resp) => match resp {
+            Ok(resp) => print_element(resp),
+            Err(e) => fatal!("An I/O error occurred while querying: {}", e),
+        }
+    }
+}
+
+fn print_element(el: Element) {
+    match el {
         Element::String(st) => write_str!(st),
         Element::Binstr(st) => write_binstr!(st),
         Element::Array(Array::Bin(brr)) => print_bin_array(brr),
@@ -71,9 +97,6 @@ impl Runner {
         Element::Array(Array::Flat(frr)) => write_flat_array(frr),
         Element::Array(Array::Recursive(a)) => print_array(a),
         _ => eskysh!("The server possibly sent a newer data type that we can't parse"),
-            },
-            Err(e) => fatal!("An I/O error occurred while querying: {}", e),
-        }
     }
 }