add connect subcommand (#57)

* first commit

* modify time depend version

* separate connect handler from origin

* read config from yaml

Author: AthLw

bin/Cargo.toml

@@ -24,3 +24,5 @@ webrtc = "0.9.0"
 serde_yaml = "0.9.30"
 notify = { version = "6.1.1", default-features = false, features = ["macos_kqueue"] }
 futures = "0.3.30"
+time = "0.3.35"
+reqwest = { version = "0.11", features = ["json"] }

bin/src/cs.rs

@@ -20,10 +20,19 @@
 #![allow(unused)]
 
 use clap::Args;
+use log::info;
 use serde::{Deserialize, Serialize};
-use std::ffi::{c_char, c_void, CString};
+use std::{ffi::{c_char, c_void, CString}, fmt::Debug, process::ExitCode};
 include!("cs_bindings.rs");
 
+use std::fs;
+use std::path::Path;
+use std::process;
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use crate::peer::*;
+
 #[derive(Args, Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
 pub struct ServerArgs {
     /// Config file path
@@ -38,6 +47,13 @@ pub struct ClientArgs {
     pub config: Option<String>,
 }
 
+#[derive(Args, Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
+pub struct ConnectArgs {
+    /// Config file path
+    #[arg(short, long)]
+    pub config: Option<String>,
+}
+
 fn convert_to_go_slices(vec: &Vec<String>) -> (GoSlice, Vec<GoString>) {
     let mut go_slices: Vec<GoString> = Vec::with_capacity(vec.len());
 
@@ -57,6 +73,85 @@ fn convert_to_go_slices(vec: &Vec<String>) -> (GoSlice, Vec<GoString>) {
         go_slices,
     )
 }
+
+fn load_config(config_path: &str) -> Result<ConnectConfig, Box<dyn std::error::Error>> {
+    // 验证文件是否存在
+    if !Path::new(config_path).exists() {
+        return Err(format!("Config file '{}' does not exist", config_path).into());
+    }
+
+    // 读取文件内容
+    let config_content = fs::read_to_string(config_path)
+        .map_err(|e| format!("Failed to read config file '{}': {}", config_path, e))?;
+
+    // 验证文件不为空
+    if config_content.trim().is_empty() {
+        return Err("Config file is empty".into());
+    }
+
+    // 解析 YAML
+    let config: ConnectConfig = serde_yaml::from_str(&config_content)
+        .map_err(|e| format!("Failed to parse YAML config: {}", e))?;
+
+    match serde_yaml::from_str::<ConnectConfig>(&config_content) {
+        Ok(config) => println!("解析成功: {:?}", config),
+        Err(e) => println!("解析错误: {}", e),
+    }
+    
+    // 验证必要的字段
+    validate_config(&config)?;
+
+    Ok(config)
+}
+
+fn validate_config(config: &ConnectConfig) -> Result<(), Box<dyn std::error::Error>> {
+    // 配置验证
+    if config.options.tcp_forward_addr.trim().is_empty() {
+        return Err("tcp_forward_addr cannot be empty".into());
+    }
+    if config.options.tcp_forward_host_prefix.trim().is_empty() {
+        return Err("tcp_forward_host_prefix cannot be empty".into());
+    }
+    Ok(())
+}
+
+
+pub fn run_connect(connect_args: ConnectArgs) {
+    let mut args = if let Some(config_path) = &connect_args.config {
+        match load_config(config_path) {
+            Ok(config) => {
+                println!("Successfully loaded config from '{}'", config_path);
+                println!("Config details:");
+                println!("  TCP Forward Address: {}", config.options.tcp_forward_addr);
+                println!("  TCP Forward Host Prefix: {}", config.options.tcp_forward_host_prefix);
+                config
+            },
+            Err(e) => {
+                eprintln!("Error loading config: {}", e);
+                process::exit(1);
+            }
+        }
+    } else {
+        println!("No config file specified, using default configuration");
+        ConnectConfig::default()
+    };
+    info!("Run connect cmd.");
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    rt.block_on(async move {
+        info!("Runtime started.");
+        let connect_reader = tokio::io::stdin();
+        let connect_writer = tokio::io::stdout();
+        // let reader = Arc::new(Mutex::new(connect_reader));
+        // let writer = Arc::new(Mutex::new(connect_writer));
+        if let Err(e) = process_connect(connect_reader, connect_writer, args).await {
+            eprintln!("process p2p connect: {}", e);
+            process::exit(1);
+        };
+    });
+    unsafe {}
+    // TODO
+}
+
 pub fn run_client(client_args: ClientArgs) {
     let mut args = if let Some(config) = client_args.config {
         vec!["client".to_owned(), "-config".to_owned(), config]

bin/src/main.rs

@@ -18,6 +18,7 @@ use std::path::PathBuf;
 
 use clap::Parser;
 use clap::Subcommand;
+use cs::ConnectArgs;
 use env_logger::Env;
 use log::{error, info};
 
@@ -50,13 +51,17 @@ enum Commands {
     Server(ServerArgs),
     /// Run GT Client
     Client(ClientArgs),
+    /// Run GT Connect
+    Connect(ConnectArgs),
 
     #[command(hide = true)]
     SubP2P,
     #[command(hide = true)]
     SubServer(ServerArgs),
     #[command(hide = true)]
     SubClient(ClientArgs),
+    #[command(hide = true)]
+    SubConnect(ConnectArgs),
 }
 
 fn main() {
@@ -75,6 +80,7 @@ fn main() {
         depth: cli.depth,
         server_args: None,
         client_args: None,
+        connect_args: None,
     };
     if let Some(command) = cli.command {
         match command {
@@ -84,6 +90,9 @@ fn main() {
             Commands::Client(args) => {
                 manager_args.client_args = Some(args);
             }
+            Commands::Connect(args) => {
+                manager_args.connect_args = Some(args);
+            }
             Commands::SubP2P => {
                 info!("GT SubP2P");
                 peer::start_peer_connection();
@@ -102,6 +111,12 @@ fn main() {
                 info!("GT SubClient done");
                 return;
             }
+            Commands::SubConnect(args) => {
+                info!("GT SubConnect");
+                cs::run_connect(args);
+                info!("GT SubConnect done");
+                return;
+            }
         }
     }
 

bin/src/manager.rs

@@ -41,14 +41,15 @@ use tokio::sync::{mpsc, Mutex, oneshot};
 use tokio::sync::oneshot::{Receiver, Sender};
 use tokio::time::timeout;
 
-use crate::cs::{ClientArgs, ServerArgs};
+use crate::cs::{ClientArgs, ConnectArgs, ServerArgs};
 
 #[derive(Debug)]
 pub struct ManagerArgs {
     pub config: Option<PathBuf>,
     pub depth: Option<u8>,
     pub server_args: Option<ServerArgs>,
     pub client_args: Option<ClientArgs>,
+    pub connect_args: Option<ConnectArgs>,
 }
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)]
@@ -71,6 +72,7 @@ enum ProcessConfigEnum {
     Config(PathBuf),
     Server(ServerArgs),
     Client(ClientArgs),
+    Connect(ConnectArgs),
 }
 
 pub struct Manager {
@@ -323,6 +325,11 @@ impl Manager {
                             $cmd.arg("-c").arg(path.clone());
                         }
                     }
+                    ProcessConfigEnum::Connect(args) => {
+                        if let Some(path) = &args.config {
+                            $cmd.arg("-c").arg(path.clone());
+                        }
+                    }
                 }
                 $cmd.stdin(Stdio::piped());
                 $cmd.stdout(Stdio::piped());
@@ -452,17 +459,21 @@ impl Manager {
     async fn run_configs(&self, configs: Vec<ProcessConfigEnum>) -> Result<()> {
         let mut server_config = vec![];
         let mut client_config = vec![];
+        let mut connect_config = vec![];
         for config in configs {
             match &config {
                 ProcessConfigEnum::Config(path) => {
                     if is_client_config_path(path).context("is_client_config_path failed")? {
                         client_config.push(config);
-                    } else {
+                    } else if is_server_config_path(path).context("is_server_config_path failed")? {
                         server_config.push(config);
+                    } else {
+                        connect_config.push(config);
                     }
                 }
                 ProcessConfigEnum::Server(_) => server_config.push(config),
                 ProcessConfigEnum::Client(_) => client_config.push(config),
+                ProcessConfigEnum::Connect(_) => connect_config.push(config),
             }
         }
         if !server_config.is_empty() {
@@ -476,6 +487,12 @@ impl Manager {
                 .await
                 .context("run_client failed")?;
         }
+
+        if !connect_config.is_empty() {
+            Self::run(self.cmds.clone(), connect_config, "sub-connect")
+                .await
+                .context("run_connect failed")?;
+        }
         Ok(())
     }
 
@@ -802,6 +819,11 @@ fn is_client_config_path(path: &PathBuf) -> Result<bool> {
     is_client_config(&yaml)
 }
 
+fn is_server_config_path(path: &PathBuf) -> Result<bool> {
+    let yaml = fs::read_to_string(path)?;
+    is_server_config(&yaml)
+}
+
 fn is_client_config(yaml: &str) -> Result<bool> {
     let c = serde_yaml::from_str::<Config>(yaml)?;
     if c.services.is_some() {
@@ -811,6 +833,23 @@ fn is_client_config(yaml: &str) -> Result<bool> {
         return match typ.as_str() {
             "client" => Ok(true),
             "server" => Ok(false),
+            "connect" => Ok(false),
+            t => Err(anyhow!("invalid config type {}", t)),
+        };
+    }
+    Ok(false)
+}
+
+fn is_server_config(yaml: &str) -> Result<bool> {
+    let s = serde_yaml::from_str::<Config>(yaml)?;
+    if s.services.is_some() {
+        return Ok(false);
+    }
+    if let Some(typ) = s.typ {
+        return match typ.as_str() {
+            "client" => Ok(false),
+            "server" => Ok(true),
+            "connect" => Ok(false),
             t => Err(anyhow!("invalid config type {}", t)),
         };
     }