From 635ded4f6815d738dd9b9b711aa4c7cf302d340b Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Sun, 12 Oct 2025 18:17:08 +0800 Subject: feat: Add connection infrastructure and documentation - Implement action service for connection handling - Add error types for connection operations - Create todo.md for project tracking --- .../vcs_actions/src/connection/action_service.rs | 129 +++++++++++++++++++++ crates/vcs_actions/src/connection/error.rs | 14 +++ 2 files changed, 143 insertions(+) create mode 100644 crates/vcs_actions/src/connection/action_service.rs create mode 100644 crates/vcs_actions/src/connection/error.rs (limited to 'crates/vcs_actions/src/connection') diff --git a/crates/vcs_actions/src/connection/action_service.rs b/crates/vcs_actions/src/connection/action_service.rs new file mode 100644 index 0000000..8d3a03d --- /dev/null +++ b/crates/vcs_actions/src/connection/action_service.rs @@ -0,0 +1,129 @@ +use std::{net::SocketAddr, path::PathBuf, sync::Arc}; + +use action_system::action_pool::ActionPool; +use cfg_file::config::ConfigFile; +use tcp_connection::{error::TcpTargetError, instance::ConnectionInstance}; +use tokio::{ + net::{TcpListener, TcpStream}, + select, signal, spawn, + sync::mpsc, +}; +use vcs_data::data::vault::{Vault, config::VaultConfig}; + +use crate::registry::server_registry::server_action_pool; + +// Start the server with a Vault using the specified directory +pub async fn server_entry(path: impl Into) -> Result<(), TcpTargetError> { + // Read the vault cfg + let vault_cfg = VaultConfig::read().await?; + + // Create TCPListener + let listener = create_tcp_listener(&vault_cfg).await?; + + // Initialize the vault + let vault: Arc = init_vault(vault_cfg, path.into()).await?; + + // Create ActionPool + let action_pool: Arc = Arc::new(server_action_pool()); + + // Start the server + let (_shutdown_rx, future) = build_server_future(vault.clone(), action_pool.clone(), listener); + let _ = future.await?; // Start and block until shutdown + + Ok(()) +} + +async fn create_tcp_listener(cfg: &VaultConfig) -> Result { + let local_bind_addr = cfg.server_config().local_bind(); + let bind_port = cfg.server_config().port(); + let sock_addr = SocketAddr::new(local_bind_addr.clone(), bind_port); + let listener = TcpListener::bind(sock_addr).await?; + + Ok(listener) +} + +async fn init_vault(cfg: VaultConfig, path: PathBuf) -> Result, TcpTargetError> { + // Init and create the vault + let Some(vault) = Vault::init(cfg, path) else { + return Err(TcpTargetError::NotFound("Vault not found".to_string())); + }; + let vault: Arc = Arc::new(vault); + + Ok(vault) +} + +fn build_server_future( + vault: Arc, + action_pool: Arc, + listener: TcpListener, +) -> ( + mpsc::Sender<()>, + impl std::future::Future>, +) { + let (tx, mut rx) = mpsc::channel::(100); + let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<()>(1); + let mut active_connections = 0; + let mut shutdown_requested = false; + + // Spawn task to handle Ctrl+C + let shutdown_tx_clone = shutdown_tx.clone(); + spawn(async move { + if let Ok(()) = signal::ctrl_c().await { + let _ = shutdown_tx_clone.send(()).await; + } + }); + + let future = async move { + loop { + select! { + // Accept new connections + accept_result = listener.accept(), if !shutdown_requested => { + match accept_result { + Ok((stream, _addr)) => { + active_connections += 1; + let _ = tx.send(1).await; + + let vault_clone = vault.clone(); + let action_pool_clone = action_pool.clone(); + let tx_clone = tx.clone(); + spawn(async move { + process_connection(stream, vault_clone, action_pool_clone).await; + let _ = tx_clone.send(-1).await; + }); + } + Err(_) => { + continue; + } + } + } + + // Handle connection count updates + Some(count_change) = rx.recv() => { + active_connections = (active_connections as i32 + count_change) as usize; + + // Check if we should shutdown after all connections are done + if shutdown_requested && active_connections == 0 { + break; + } + } + + // Handle shutdown signal + _ = shutdown_rx.recv() => { + shutdown_requested = true; + // If no active connections, break immediately + if active_connections == 0 { + break; + } + } + } + } + + Ok(()) + }; + + (shutdown_tx, future) +} + +async fn process_connection(stream: TcpStream, vault: Arc, action_pool: Arc) { + let instance = ConnectionInstance::from(stream); +} diff --git a/crates/vcs_actions/src/connection/error.rs b/crates/vcs_actions/src/connection/error.rs new file mode 100644 index 0000000..241c16e --- /dev/null +++ b/crates/vcs_actions/src/connection/error.rs @@ -0,0 +1,14 @@ +use std::io; +use thiserror::Error; + +#[derive(Error, Debug, Clone)] +pub enum ConnectionError { + #[error("I/O error: {0}")] + Io(String), +} + +impl From for ConnectionError { + fn from(error: io::Error) -> Self { + ConnectionError::Io(error.to_string()) + } +} -- cgit From 67fb8ec01b351c6c9fd2af321166bb92250b1218 Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 13 Oct 2025 13:34:39 +0800 Subject: feat: Implement JSON-based type-erased action invocation - Add process_json method to ActionPool for type-agnostic calls using JSON serialization - Extend ActionContext with action_name and action_args fields and setter methods - Update action_gen macro to use process_json instead of typed process method - Implement remote action invocation framework in client_registry and action_service - Add protocol definitions for remote action communication - Enable flexible action execution without explicit type specifications --- .../vcs_actions/src/connection/action_service.rs | 29 ++++++++++++++++++---- crates/vcs_actions/src/connection/protocol.rs | 7 ++++++ 2 files changed, 31 insertions(+), 5 deletions(-) create mode 100644 crates/vcs_actions/src/connection/protocol.rs (limited to 'crates/vcs_actions/src/connection') diff --git a/crates/vcs_actions/src/connection/action_service.rs b/crates/vcs_actions/src/connection/action_service.rs index 8d3a03d..9ea5957 100644 --- a/crates/vcs_actions/src/connection/action_service.rs +++ b/crates/vcs_actions/src/connection/action_service.rs @@ -1,6 +1,6 @@ use std::{net::SocketAddr, path::PathBuf, sync::Arc}; -use action_system::action_pool::ActionPool; +use action_system::{action::ActionContext, action_pool::ActionPool}; use cfg_file::config::ConfigFile; use tcp_connection::{error::TcpTargetError, instance::ConnectionInstance}; use tokio::{ @@ -10,10 +10,12 @@ use tokio::{ }; use vcs_data::data::vault::{Vault, config::VaultConfig}; -use crate::registry::server_registry::server_action_pool; +use crate::{ + connection::protocol::RemoteActionInvoke, registry::server_registry::server_action_pool, +}; // Start the server with a Vault using the specified directory -pub async fn server_entry(path: impl Into) -> Result<(), TcpTargetError> { +pub async fn server_entry(vault_path: impl Into) -> Result<(), TcpTargetError> { // Read the vault cfg let vault_cfg = VaultConfig::read().await?; @@ -21,7 +23,7 @@ pub async fn server_entry(path: impl Into) -> Result<(), TcpTargetError let listener = create_tcp_listener(&vault_cfg).await?; // Initialize the vault - let vault: Arc = init_vault(vault_cfg, path.into()).await?; + let vault: Arc = init_vault(vault_cfg, vault_path.into()).await?; // Create ActionPool let action_pool: Arc = Arc::new(server_action_pool()); @@ -125,5 +127,22 @@ fn build_server_future( } async fn process_connection(stream: TcpStream, vault: Arc, action_pool: Arc) { - let instance = ConnectionInstance::from(stream); + // Setup connection instance + let mut instance = ConnectionInstance::from(stream); + + // Read action name and action arguments + let Ok(msg) = instance.read_msgpack::().await else { + return; + }; + + // Build context + let ctx = ActionContext::remote().insert_instance(instance); + + // Process action + let Ok(_result_json) = action_pool + .process_json(&msg.action_name, ctx, msg.action_args_json) + .await + else { + return; + }; } diff --git a/crates/vcs_actions/src/connection/protocol.rs b/crates/vcs_actions/src/connection/protocol.rs new file mode 100644 index 0000000..2cebe79 --- /dev/null +++ b/crates/vcs_actions/src/connection/protocol.rs @@ -0,0 +1,7 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Default, Clone, Serialize, Deserialize)] +pub struct RemoteActionInvoke { + pub action_name: String, + pub action_args_json: String, +} -- cgit From acf0804b5f9bdc2796d847919a8ae20103be600a Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 13 Oct 2025 14:17:51 +0800 Subject: feat: implement asynchronous action call system - Add async callback support with proper argument passing - Implement remote action invocation via TCP connection - Add hello_world_action example demonstrating async communication - Improve ActionPool with type-safe async processing - Update client registry for remote action handling - Enhance ActionContext with better instance management - Support both local and remote action execution modes --- crates/vcs_actions/src/connection/action_service.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) (limited to 'crates/vcs_actions/src/connection') diff --git a/crates/vcs_actions/src/connection/action_service.rs b/crates/vcs_actions/src/connection/action_service.rs index 9ea5957..0a49953 100644 --- a/crates/vcs_actions/src/connection/action_service.rs +++ b/crates/vcs_actions/src/connection/action_service.rs @@ -136,7 +136,10 @@ async fn process_connection(stream: TcpStream, vault: Arc, action_pool: A }; // Build context - let ctx = ActionContext::remote().insert_instance(instance); + let ctx: ActionContext = ActionContext::remote().insert_instance(instance); + + // Insert vault into context + let ctx = ctx.insert_arc(vault); // Process action let Ok(_result_json) = action_pool -- cgit From 4810f56e6a49b60923eb850d5944457650c81c75 Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 13 Oct 2025 14:27:01 +0800 Subject: Fix Clippy warnings and optimize code - Fix let_underscore_future warning by properly awaiting async functions - Make accept_import function async to match add_mapping usage - Propagate errors properly with ? operator instead of ignoring them - Replace manual Default implementation with derive attribute - Replace vec! with array literal to avoid useless_vec warning - All tests pass and code is now Clippy clean --- crates/vcs_actions/src/connection/action_service.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'crates/vcs_actions/src/connection') diff --git a/crates/vcs_actions/src/connection/action_service.rs b/crates/vcs_actions/src/connection/action_service.rs index 0a49953..c302fd4 100644 --- a/crates/vcs_actions/src/connection/action_service.rs +++ b/crates/vcs_actions/src/connection/action_service.rs @@ -30,7 +30,7 @@ pub async fn server_entry(vault_path: impl Into) -> Result<(), TcpTarge // Start the server let (_shutdown_rx, future) = build_server_future(vault.clone(), action_pool.clone(), listener); - let _ = future.await?; // Start and block until shutdown + future.await?; // Start and block until shutdown Ok(()) } @@ -38,7 +38,7 @@ pub async fn server_entry(vault_path: impl Into) -> Result<(), TcpTarge async fn create_tcp_listener(cfg: &VaultConfig) -> Result { let local_bind_addr = cfg.server_config().local_bind(); let bind_port = cfg.server_config().port(); - let sock_addr = SocketAddr::new(local_bind_addr.clone(), bind_port); + let sock_addr = SocketAddr::new(*local_bind_addr, bind_port); let listener = TcpListener::bind(sock_addr).await?; Ok(listener) -- cgit