update cli

Author: luoluoyuyu

cli/cli/Cargo.toml

@@ -16,7 +16,7 @@ function-stream = { path = "../../" }
 protocol = { path = "../../protocol" }
 clap = { version = "4.5", features = ["derive"] }
 thiserror = "2"
-tokio = { version = "1.0", features = ["full"] }
+tokio = { version = "1.0", features = ["full", "signal"] }
 tonic = { version = "0.12", features = ["default"] }
 rustyline = { version = "14.0", features = ["with-dirs"] }
 

cli/cli/src/main.rs

@@ -15,6 +15,8 @@ mod repl;
 use clap::Parser;
 use repl::Repl;
 use std::process;
+use std::sync::Arc;
+use tokio::sync::Mutex;
 
 #[derive(Parser, Debug)]
 #[command(name = "function-stream-cli")]
@@ -35,9 +37,9 @@ struct Args {
 async fn main() {
     let args = Args::parse();
 
-    let mut repl = Repl::new(args.host.clone(), args.port);
+    let repl = Arc::new(Mutex::new(Repl::new(args.host.clone(), args.port)));
 
-    if let Err(e) = repl.run_async().await {
+    if let Err(e) = Repl::run_async(repl).await {
         eprintln!("Error: {}", e);
         process::exit(1);
     }

cli/cli/src/repl.rs

@@ -21,6 +21,8 @@ use protocol::cli::{function_stream_service_client::FunctionStreamServiceClient,
 use rustyline::error::ReadlineError;
 use rustyline::{Config, DefaultEditor, EditMode};
 use std::io::{self, Cursor, Write};
+use std::sync::Arc;
+use tokio::sync::Mutex;
 use tonic::Request;
 
 #[derive(Debug, thiserror::Error)]
@@ -269,22 +271,81 @@ impl Repl {
         }
     }
 
-    pub async fn run_async(&mut self) -> io::Result<()> {
-        println!("Function Stream SQL Interface");
-        println!("Server: {}\n", self.server_address());
-
-        if let Err(e) = self.connect().await {
-            eprintln!("Error: {}", e);
-            return Ok(());
+    pub async fn run_async(repl: Arc<Mutex<Self>>) -> io::Result<()> {
+        {
+            let mut guard = repl.lock().await;
+            println!("Function Stream SQL Interface");
+            println!("Server: {}\n", guard.server_address());
+            if let Err(e) = guard.connect().await {
+                eprintln!("Error: {}", e);
+                return Ok(());
+            }
         }
 
+        #[cfg(unix)]
+        let mut sigterm = {
+            use tokio::signal::unix::{signal, SignalKind};
+            signal(SignalKind::terminate()).expect("failed to register SIGTERM handler")
+        };
+
+        let mut skip_save_history = false;
         loop {
-            let input = match self.read_sql_input() {
-                Ok(sql) => sql,
-                Err(ReadlineError::Interrupted) => continue,
-                Err(ReadlineError::Eof) => break,
-                Err(e) => {
+            let repl_clone = repl.clone();
+            let read_result = {
+                #[cfg(unix)]
+                {
+                    tokio::select! {
+                        _ = tokio::signal::ctrl_c() => None,
+                        _ = sigterm.recv() => None,
+                        result = tokio::task::spawn_blocking(move || {
+                            let mut guard = repl_clone.blocking_lock();
+                            guard.read_sql_input()
+                        }) => match result {
+                            Ok(Ok(sql)) => Some(Ok(sql)),
+                            Ok(Err(e)) => Some(Err(e)),
+                            Err(e) => {
+                                eprintln!("Read Error: {}", e);
+                                None
+                            }
+                        },
+                    }
+                }
+                #[cfg(not(unix))]
+                {
+                    tokio::select! {
+                        _ = tokio::signal::ctrl_c() => None,
+                        result = tokio::task::spawn_blocking(move || {
+                            let mut guard = repl_clone.blocking_lock();
+                            guard.read_sql_input()
+                        }) => match result {
+                            Ok(Ok(sql)) => Some(Ok(sql)),
+                            Ok(Err(e)) => Some(Err(e)),
+                            Err(e) => {
+                                eprintln!("Read Error: {}", e);
+                                None
+                            }
+                        },
+                    }
+                }
+            };
+
+            let input = match read_result {
+                None => {
+                    skip_save_history = true;
+                    break;
+                }
+                Some(Ok(sql)) => sql,
+                Some(Err(ReadlineError::Interrupted)) => {
+                    skip_save_history = true;
+                    break;
+                }
+                Some(Err(ReadlineError::Eof)) => {
+                    skip_save_history = true;
+                    break;
+                }
+                Some(Err(e)) => {
                     eprintln!("Read Error: {}", e);
+                    skip_save_history = true;
                     break;
                 }
             };
@@ -293,28 +354,38 @@ impl Repl {
                 continue;
             }
 
+            let mut guard = repl.lock().await;
             match input.trim().to_lowercase().as_str() {
                 "exit" | "quit" | "q" => break,
-                "help" | "h" => self.print_help(),
+                "help" | "h" => guard.print_help(),
                 _ => {
-                    if let Err(e) = self.execute_sql(&input).await {
+                    if let Err(e) = guard.execute_sql(&input).await {
                         eprintln!("SQL Execution Error: {}", e);
+                    } else {
+                        guard.add_history_entry(&input);
                     }
                 }
             }
+            drop(guard);
             println!();
         }
 
-        if let Some(ref mut ed) = self.editor {
-            let _ = ed.save_history(".function-stream-cli-history");
+        if !skip_save_history {
+            if let Some(ref mut ed) = repl.lock().await.editor {
+                let _ = ed.save_history(".function-stream-cli-history");
+            }
         }
         Ok(())
     }
 
     fn read_sql_input(&mut self) -> Result<String, ReadlineError> {
         let mut lines = Vec::new();
         loop {
-            let prompt = if lines.is_empty() { "sql> " } else { "  -> " };
+            let prompt = if lines.is_empty() {
+                "function-stream> "
+            } else {
+                "  -> "
+            };
 
             let line = match self.editor.as_mut() {
                 Some(ed) => ed.readline(prompt)?,
@@ -332,11 +403,6 @@ impl Repl {
             let trimmed = lines.last().map(|s| s.trim()).unwrap_or("");
 
             if trimmed.ends_with(';') || self.is_balanced(&full_sql) {
-                if let Some(ed) = self.editor.as_mut() {
-                    if !full_sql.is_empty() {
-                        let _ = ed.add_history_entry(full_sql.as_str());
-                    }
-                }
                 return Ok(full_sql);
             }
         }
@@ -348,6 +414,14 @@ impl Repl {
         open == close && !sql.trim().is_empty()
     }
 
+    fn add_history_entry(&mut self, entry: &str) {
+        if let Some(ed) = self.editor.as_mut() {
+            if !entry.trim().is_empty() {
+                let _ = ed.add_history_entry(entry.trim());
+            }
+        }
+    }
+
     fn print_help(&self) {
         let mut table = Table::new();
         table

docs/sql-cli-guide-zh.md

@@ -53,7 +53,7 @@ SQL CLI 通过 gRPC 协议与远程 Server 通信。
 cargo run -p function-stream-cli -- -h <SERVER_HOST> -p <SERVER_PORT>
 ```
 
-- **终端提示符**：成功连接后显示 `sql>`。
+- **终端提示符**：成功连接后显示 `function-stream>`。
 - **输入规范**：支持多行输入，系统通过 `;`（分号）或括号平衡检测来判定语句结束并提交执行。
 
 ---
@@ -131,7 +131,7 @@ DROP FUNCTION go_processor_demo;
 
 ## 三、REPL 内建辅助指令
 
-在 `sql>` 提示符下，除了标准 SQL 语句，还支持以下便捷指令：
+在 `function-stream>` 提示符下，支持以下便捷指令：
 
 | 指令    | 简写  | 说明                              |
 |-------|-----|---------------------------------|

docs/sql-cli-guide.md

@@ -53,7 +53,7 @@ SQL CLI communicates with the remote Server via the gRPC protocol.
 cargo run -p function-stream-cli -- -h <SERVER_HOST> -p <SERVER_PORT>
 ```
 
-- **Terminal Prompt**: Displays `sql>` after successful connection.
+- **Terminal Prompt**: Displays `function-stream>` after successful connection.
 - **Input Specification**: Supports multi-line input; the system detects statement completion and submits for execution via `;` (semicolon) or parenthesis balancing.
 
 ---
@@ -131,7 +131,7 @@ DROP FUNCTION go_processor_demo;
 
 ## 3. REPL Built-in Auxiliary Commands
 
-At the `sql>` prompt, in addition to standard SQL statements, the following convenient commands are also supported:
+At the `function-stream>` prompt, the following convenient commands are supported:
 
 | Command | Short | Description                                                               |
 |---------|-------|---------------------------------------------------------------------------|