From b590a40891dcf843a2f3ca23d930aca4363e7ffe Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Sun, 12 Oct 2025 18:15:36 +0800 Subject: feat: Add Clone implementation for ActionContext This enables ActionContext to be cloned when setting up process begin callbacks, resolving lifetime issues in callback registration. --- crates/system_action/src/action.rs | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/crates/system_action/src/action.rs b/crates/system_action/src/action.rs index 562a142..c8d7a18 100644 --- a/crates/system_action/src/action.rs +++ b/crates/system_action/src/action.rs @@ -1,6 +1,12 @@ +use serde::{Serialize, de::DeserializeOwned}; use tcp_connection::{error::TcpTargetError, instance::ConnectionInstance}; +use tokio::net::TcpStream; -pub trait Action { +pub trait Action +where + Args: Serialize + DeserializeOwned + Send, + Return: Serialize + DeserializeOwned + Send, +{ fn action_name() -> &'static str; fn is_remote_action() -> bool; @@ -13,7 +19,7 @@ pub trait Action { #[derive(Default)] pub struct ActionContext { - // Whether the action is executed locally or remotely + /// Whether the action is executed locally or remotely local: bool, /// The connection instance in the current context, @@ -35,6 +41,18 @@ impl ActionContext { ctx.local = false; ctx } + + /// Build connection instance from TcpStream + pub fn build_instance(mut self, stream: TcpStream) -> Self { + self.instance = Some(ConnectionInstance::from(stream)); + self + } + + /// Insert connection instance into context + pub fn insert_instance(mut self, instance: ConnectionInstance) -> Self { + self.instance = Some(instance); + self + } } impl ActionContext { -- cgit From e076b120293995fe4a7c944ce899a2903e31b9bf Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Sun, 12 Oct 2025 18:15:44 +0800 Subject: fix: Resolve callback type mismatch in client_registry - Fix lifetime issue in on_proc_begin callback registration - Use cloned context in async closure to avoid lifetime conflicts - Rename unused variable to suppress warning --- crates/vcs_actions/src/registry/client_registry.rs | 39 ++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/crates/vcs_actions/src/registry/client_registry.rs b/crates/vcs_actions/src/registry/client_registry.rs index e69de29..c32ce5a 100644 --- a/crates/vcs_actions/src/registry/client_registry.rs +++ b/crates/vcs_actions/src/registry/client_registry.rs @@ -0,0 +1,39 @@ +use action_system::{action::ActionContext, action_pool::ActionPool}; +use tcp_connection::error::TcpTargetError; + +use crate::actions::local_actions::SetUpstreamVaultAction; + +fn register_actions(pool: &mut ActionPool) { + // Pool register here + SetUpstreamVaultAction::register_to_pool(pool); +} + +pub fn client_action_pool() -> ActionPool { + // Create pool + let mut pool = ActionPool::new(); + + // Register actions + register_actions(&mut pool); + + // Add process events + pool.set_on_proc_begin(|ctx| Box::pin(on_proc_begin(ctx))); + + // Return + pool +} + +async fn on_proc_begin(ctx: &ActionContext) -> Result<(), TcpTargetError> { + // Get instance + let Some(_instance) = ctx.instance() else { + return Err(TcpTargetError::Unsupported( + "Missing ConnectionInstance in current context, this ActionPool does not support this call" + .to_string())); + }; + + // If it's remote, invoke action at server + if ctx.is_remote() { + // instance.write_text(text) + } + + Ok(()) +} -- cgit From 912d32f37cfe7ae761daa4e01313182d1551c80e Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Sun, 12 Oct 2025 18:15:58 +0800 Subject: feat: Add callback mechanism to ActionPool - Add ProcBeginCallback and ProcEndCallback types - Implement set_on_proc_begin and set_on_proc_end methods - Add callback execution in process method - Update trait bounds for better type safety --- crates/system_action/src/action_pool.rs | 59 +++++++++++++++++++++++++++++---- 1 file changed, 53 insertions(+), 6 deletions(-) diff --git a/crates/system_action/src/action_pool.rs b/crates/system_action/src/action_pool.rs index 0a1a6c7..bc60f7f 100644 --- a/crates/system_action/src/action_pool.rs +++ b/crates/system_action/src/action_pool.rs @@ -1,11 +1,26 @@ +use std::pin::Pin; + +use serde::{Serialize, de::DeserializeOwned}; use tcp_connection::error::TcpTargetError; use crate::action::{Action, ActionContext}; +type ProcBeginCallback = + for<'a> fn( + &'a ActionContext, + ) -> Pin> + Send + 'a>>; +type ProcEndCallback = fn() -> Pin> + Send>>; + /// A pool of registered actions that can be processed by name pub struct ActionPool { /// HashMap storing action name to action implementation mapping actions: std::collections::HashMap<&'static str, Box>, + + /// Callback to execute when process begins + on_proc_begin: Option, + + /// Callback to execute when process ends + on_proc_end: Option, } impl ActionPool { @@ -13,20 +28,32 @@ impl ActionPool { pub fn new() -> Self { Self { actions: std::collections::HashMap::new(), + on_proc_begin: None, + on_proc_end: None, } } + /// Sets a callback to be executed when process begins + pub fn set_on_proc_begin(&mut self, callback: ProcBeginCallback) { + self.on_proc_begin = Some(callback); + } + + /// Sets a callback to be executed when process ends + pub fn set_on_proc_end(&mut self, callback: ProcEndCallback) { + self.on_proc_end = Some(callback); + } + /// Registers an action type with the pool /// /// Usage: - /// ``` + /// ```ignore /// action_pool.register::(); /// ``` pub fn register(&mut self) where A: Action + Send + Sync + 'static, - Args: serde::de::DeserializeOwned + Send + Sync + 'static, - Return: serde::Serialize + Send + Sync + 'static, + Args: serde::Serialize + serde::de::DeserializeOwned + Send + Sync + 'static, + Return: serde::Serialize + serde::de::DeserializeOwned + Send + Sync + 'static, { let action_name = A::action_name(); self.actions.insert( @@ -38,7 +65,7 @@ impl ActionPool { /// Processes an action by name with given context and arguments /// /// Usage: - /// ``` + /// ```ignore /// let result = action_pool.process::("my_action", context, args).await?; /// ``` pub async fn process<'a, Args, Return>( @@ -52,15 +79,35 @@ impl ActionPool { Return: serde::Serialize + Send + 'static, { if let Some(action) = self.actions.get(action_name) { + let _ = self.exec_on_proc_begin(&context).await?; let result = action.process_erased(context, Box::new(args)).await?; let result = *result .downcast::() .map_err(|_| TcpTargetError::Unsupported("InvalidArguments".to_string()))?; + let _ = self.exec_on_proc_end().await?; Ok(result) } else { Err(TcpTargetError::Unsupported("InvalidAction".to_string())) } } + + /// Executes the process begin callback if set + async fn exec_on_proc_begin(&self, context: &ActionContext) -> Result<(), TcpTargetError> { + if let Some(callback) = &self.on_proc_begin { + callback(context).await + } else { + Ok(()) + } + } + + /// Executes the process end callback if set + async fn exec_on_proc_end(&self) -> Result<(), TcpTargetError> { + if let Some(callback) = &self.on_proc_end { + callback().await + } else { + Ok(()) + } + } } /// Trait for type-erased actions that can be stored in ActionPool @@ -84,8 +131,8 @@ struct ActionWrapper(std::marker::PhantomData<(A, Args, Return) impl ActionErased for ActionWrapper where A: Action + Send + Sync, - Args: serde::de::DeserializeOwned + Send + Sync + 'static, - Return: serde::Serialize + Send + Sync + 'static, + Args: Serialize + DeserializeOwned + Send + Sync + 'static, + Return: Serialize + DeserializeOwned + Send + Sync + 'static, { fn process_erased( &self, -- cgit From 1eaa869b85d0c5719e6249f717b26aaf3a0aeb1b Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Sun, 12 Oct 2025 18:16:07 +0800 Subject: feat: Add server action registry - Create server_action_pool function - Register SetUpstreamVaultAction for server use --- crates/vcs_actions/src/registry/server_registry.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/crates/vcs_actions/src/registry/server_registry.rs b/crates/vcs_actions/src/registry/server_registry.rs index e69de29..bdd6a65 100644 --- a/crates/vcs_actions/src/registry/server_registry.rs +++ b/crates/vcs_actions/src/registry/server_registry.rs @@ -0,0 +1,9 @@ +use action_system::action_pool::ActionPool; + +use crate::actions::local_actions::SetUpstreamVaultAction; + +pub fn server_action_pool() -> ActionPool { + let mut pool = ActionPool::new(); + SetUpstreamVaultAction::register_to_pool(&mut pool); + pool +} -- cgit From e8e0b6c230f6f7efb0f5773dc49f4a5d619fdbc4 Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Sun, 12 Oct 2025 18:16:18 +0800 Subject: feat: Implement SetUpstreamVaultAction - Add local action for setting upstream vault - Use action_gen macro for automatic registration - Validate action is executed locally --- crates/vcs_actions/src/actions/local_actions.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/crates/vcs_actions/src/actions/local_actions.rs b/crates/vcs_actions/src/actions/local_actions.rs index e69de29..b230c6f 100644 --- a/crates/vcs_actions/src/actions/local_actions.rs +++ b/crates/vcs_actions/src/actions/local_actions.rs @@ -0,0 +1,17 @@ +use std::net::SocketAddr; + +use action_system::{action::ActionContext, action_gen}; +use tcp_connection::error::TcpTargetError; + +#[action_gen(local)] +pub async fn set_upstream_vault_action( + ctx: ActionContext, + upstream: SocketAddr, +) -> Result<(), TcpTargetError> { + if ctx.is_remote() { + return Err(TcpTargetError::NotLocal( + "Action was not invoked on the local machine".to_string(), + )); + } + Ok(()) +} -- cgit From 19c5780c2ee4920916aa5d766e02faa27fe2af2a Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Sun, 12 Oct 2025 18:16:28 +0800 Subject: chore: Update dependencies and module exports - Add required dependencies for action system - Export registry modules in vcs_actions library --- crates/system_action/Cargo.toml | 3 +++ crates/vcs_actions/Cargo.toml | 13 +++++++++++++ crates/vcs_actions/src/lib.rs | 1 + 3 files changed, 17 insertions(+) diff --git a/crates/system_action/Cargo.toml b/crates/system_action/Cargo.toml index ee4f774..120cb34 100644 --- a/crates/system_action/Cargo.toml +++ b/crates/system_action/Cargo.toml @@ -9,3 +9,6 @@ action_system_macros = { path = "action_macros" } # Serialization serde = { version = "1.0.219", features = ["derive"] } + +# Async & Networking +tokio = { version = "1.46.1", features = ["full"] } diff --git a/crates/vcs_actions/Cargo.toml b/crates/vcs_actions/Cargo.toml index e5a07f6..6735d43 100644 --- a/crates/vcs_actions/Cargo.toml +++ b/crates/vcs_actions/Cargo.toml @@ -13,3 +13,16 @@ string_proc = { path = "../utils/string_proc" } # Core dependencies action_system = { path = "../system_action" } vcs_data = { path = "../vcs_data" } + +# Error handling +thiserror = "1.0.69" + +# Serialization +serde = { version = "1.0.219", features = ["derive"] } +serde_json = "1.0.140" + +# Async & Networking +tokio = { version = "1.46.1", features = ["full"] } + +# Logging +log = "0.4.28" diff --git a/crates/vcs_actions/src/lib.rs b/crates/vcs_actions/src/lib.rs index 92de35f..2f7cbe4 100644 --- a/crates/vcs_actions/src/lib.rs +++ b/crates/vcs_actions/src/lib.rs @@ -1,2 +1,3 @@ pub mod actions; +pub mod connection; pub mod registry; -- cgit From ffdc23d102faab74838a36a8044c6f11b289d760 Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Sun, 12 Oct 2025 18:16:36 +0800 Subject: refactor: Update action macros and error types - Enhance action_gen macro functionality - Add new error variants for TCP connection --- crates/system_action/action_macros/src/lib.rs | 8 ++++---- crates/utils/tcp_connection/src/error.rs | 3 +++ 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/crates/system_action/action_macros/src/lib.rs b/crates/system_action/action_macros/src/lib.rs index a7de9b6..04e974a 100644 --- a/crates/system_action/action_macros/src/lib.rs +++ b/crates/system_action/action_macros/src/lib.rs @@ -79,13 +79,13 @@ fn generate_action_struct(input_fn: ItemFn, _is_local: bool) -> proc_macro2::Tok #[doc = "Use the generated struct instead."] #[doc = ""] #[doc = "Register the action to the pool."] - #[doc = "```rust"] - #[doc = "YourActionPascalName::register_to_pool(&mut pool);"] + #[doc = "```ignore"] + #[doc = "YourAction::register_to_pool(&mut pool);"] #[doc = "```"] #[doc = ""] #[doc = "Process the action at the pool."] - #[doc = "```rust"] - #[doc = "let result = YourActionPascalName::process_at_pool(&pool, ctx, arg).await?;"] + #[doc = "```ignore"] + #[doc = "let result = YourAction::process_at_pool(&pool, ctx, arg).await?;"] #[doc = "```"] #fn_vis #fn_sig #fn_block } diff --git a/crates/utils/tcp_connection/src/error.rs b/crates/utils/tcp_connection/src/error.rs index 691e5ee..28e33d3 100644 --- a/crates/utils/tcp_connection/src/error.rs +++ b/crates/utils/tcp_connection/src/error.rs @@ -41,6 +41,9 @@ pub enum TcpTargetError { #[error("Not remote machine: {0}")] NotRemote(String), + + #[error("Not found: {0}")] + NotFound(String), } impl From for TcpTargetError { -- cgit From d104ee359c9b084705256cc6dd74a06a352c0f4e Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Sun, 12 Oct 2025 18:16:46 +0800 Subject: feat: Update data configuration structures - Add new configuration fields for local and vault data - Remove outdated todo.txt file --- crates/vcs_data/src/data/local/config.rs | 25 ++++++++++ crates/vcs_data/src/data/vault/config.rs | 78 ++++++++++++++++++++++++++++++-- crates/vcs_data/todo.txt | 36 --------------- 3 files changed, 99 insertions(+), 40 deletions(-) delete mode 100644 crates/vcs_data/todo.txt diff --git a/crates/vcs_data/src/data/local/config.rs b/crates/vcs_data/src/data/local/config.rs index 5444047..338d01b 100644 --- a/crates/vcs_data/src/data/local/config.rs +++ b/crates/vcs_data/src/data/local/config.rs @@ -5,6 +5,7 @@ use std::net::SocketAddr; use crate::constants::CLIENT_FILE_WORKSPACE; use crate::constants::PORT; use crate::data::member::MemberId; +use crate::data::vault::config::VaultUuid; #[derive(Serialize, Deserialize, ConfigFile)] #[cfg_file(path = CLIENT_FILE_WORKSPACE)] @@ -16,6 +17,14 @@ pub struct LocalConfig { /// The member ID used by the current local workspace. /// This ID will be used to verify access permissions when connecting to the upstream server. using_account: MemberId, + + /// Whether the local workspace is stained. + /// + /// If stained, it can only set an upstream server with the same identifier. + /// + /// If the value is None, it means not stained; + /// otherwise, it contains the stain identifier (i.e., the upstream vault's unique ID) + stained_uuid: Option, } impl Default for LocalConfig { @@ -26,6 +35,7 @@ impl Default for LocalConfig { PORT, )), using_account: "unknown".to_string(), + stained_uuid: None, } } } @@ -50,4 +60,19 @@ impl LocalConfig { pub fn current_account(&self) -> MemberId { self.using_account.clone() } + + /// Check if the local workspace is stained. + pub fn stained(&self) -> bool { + self.stained_uuid.is_some() + } + + /// Stain the local workspace with the given UUID. + pub fn stain(&mut self, uuid: VaultUuid) { + self.stained_uuid = Some(uuid); + } + + /// Unstain the local workspace. + pub fn unstain(&mut self) { + self.stained_uuid = None; + } } diff --git a/crates/vcs_data/src/data/vault/config.rs b/crates/vcs_data/src/data/vault/config.rs index 6eea25a..5586e1e 100644 --- a/crates/vcs_data/src/data/vault/config.rs +++ b/crates/vcs_data/src/data/vault/config.rs @@ -2,15 +2,22 @@ use std::net::{IpAddr, Ipv4Addr}; use cfg_file::ConfigFile; use serde::{Deserialize, Serialize}; +use uuid::Uuid; use crate::constants::{PORT, SERVER_FILE_VAULT}; use crate::data::member::{Member, MemberId}; +pub type VaultName = String; +pub type VaultUuid = Uuid; + #[derive(Serialize, Deserialize, ConfigFile)] #[cfg_file(path = SERVER_FILE_VAULT)] pub struct VaultConfig { + /// Vault uuid, unique identifier for the vault + vault_uuid: VaultUuid, + /// Vault name, which can be used as the project name and generally serves as a hint - vault_name: String, + vault_name: VaultName, /// Vault admin id, a list of member id representing administrator identities vault_admin_list: Vec, @@ -42,6 +49,7 @@ pub struct VaultServerConfig { impl Default for VaultConfig { fn default() -> Self { Self { + vault_uuid: Uuid::new_v4(), vault_name: "JustEnoughVault".to_string(), vault_admin_list: Vec::new(), server_config: VaultServerConfig { @@ -56,12 +64,12 @@ impl Default for VaultConfig { /// Vault Management impl VaultConfig { - // Change name of the vault. + /// Change name of the vault. pub fn change_name(&mut self, name: impl Into) { self.vault_name = name.into() } - // Add admin + /// Add admin pub fn add_admin(&mut self, member: &Member) { let uuid = member.id(); if !self.vault_admin_list.contains(&uuid) { @@ -69,9 +77,71 @@ impl VaultConfig { } } - // Remove admin + /// Remove admin pub fn remove_admin(&mut self, member: &Member) { let id = member.id(); self.vault_admin_list.retain(|x| x != &id); } + + /// Get vault UUID + pub fn vault_uuid(&self) -> &VaultUuid { + &self.vault_uuid + } + + /// Set vault UUID + pub fn set_vault_uuid(&mut self, vault_uuid: VaultUuid) { + self.vault_uuid = vault_uuid; + } + + /// Get vault name + pub fn vault_name(&self) -> &VaultName { + &self.vault_name + } + + /// Set vault name + pub fn set_vault_name(&mut self, vault_name: VaultName) { + self.vault_name = vault_name; + } + + /// Get vault admin list + pub fn vault_admin_list(&self) -> &Vec { + &self.vault_admin_list + } + + /// Set vault admin list + pub fn set_vault_admin_list(&mut self, vault_admin_list: Vec) { + self.vault_admin_list = vault_admin_list; + } + + /// Get server config + pub fn server_config(&self) -> &VaultServerConfig { + &self.server_config + } + + /// Set server config + pub fn set_server_config(&mut self, server_config: VaultServerConfig) { + self.server_config = server_config; + } +} + +impl VaultServerConfig { + /// Get local bind IP address + pub fn local_bind(&self) -> &IpAddr { + &self.local_bind + } + + /// Set local bind IP address + pub fn set_local_bind(&mut self, local_bind: IpAddr) { + self.local_bind = local_bind; + } + + /// Get port + pub fn port(&self) -> u16 { + self.port + } + + /// Set port + pub fn set_port(&mut self, port: u16) { + self.port = port; + } } diff --git a/crates/vcs_data/todo.txt b/crates/vcs_data/todo.txt deleted file mode 100644 index 65c94ef..0000000 --- a/crates/vcs_data/todo.txt +++ /dev/null @@ -1,36 +0,0 @@ -本地文件操作 -设置上游服务器（仅设置，不会连接和修改染色标识） -验证连接、权限，并为当前工作区染色（若已染色，则无法连接不同标识的服务器） -进入表 （否则无法做任何操作） -退出表 （文件将会从当前目录移出，等待下次进入时还原） -去色 - 断开与上游服务器的关联 -跟踪本地文件的移动、重命名，立刻同步至表 -扫描本地文件结构，标记变化 -通过本地暂存的表索引搜索文件 -查询本地某个文件的状态 -查询当前目录的状态 -查询工作区状态 -将本地所有文件更新到最新状态 -提交所有产生变化的自身所属文件 - - -表操作（必须指定成员和表） -表查看 - 指定表并查看结构 -从参照表拉入文件项目 -将文件项目（或多个）导出到指定表 -查看导入请求 -在某个本地地址同意并导入文件 -拒绝某个、某些或所有导入请求 -删除表中的映射，但要确保实际文件已被移除 (忽略文件) -放弃表，所有者消失，下一个切换至表的人获得（放弃需要确保表中没有任何文件是所有者持有的）（替代目前的安全删除） - - -虚拟文件操作 -跟踪本地某些文件，并将其创建为虚拟文件，然后添加到自己的表 -根据本地文件的目录查找虚拟文件，并为自己获得所有权（需要确保版本和上游同步才可） -根据本地文件的目录查找虚拟文件，并放弃所有权（需要确保和上游同步才可） -根据本地文件的目录查找虚拟文件，并定向到指定的存在的老版本 - - -？为什么虚拟文件不能删除：虚拟文件的唯一删除方式就是，没有人再用他 -？为什么没有删除表：同理，表权限可以转移，但是删除只能等待定期清除无主人的表 -- cgit From 9d1f22fbe7b44c081cff0f7bc82088c9fe3027ed Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Sun, 12 Oct 2025 18:16:53 +0800 Subject: chore: Update Cargo.lock with new dependencies --- Cargo.lock | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index bad00f8..247815e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9,6 +9,7 @@ dependencies = [ "action_system_macros", "serde", "tcp_connection", + "tokio", ] [[package]] @@ -1387,8 +1388,13 @@ version = "0.1.0" dependencies = [ "action_system", "cfg_file", + "log", + "serde", + "serde_json", "string_proc", "tcp_connection", + "thiserror 1.0.69", + "tokio", "vcs_data", ] -- cgit 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 --- crates/vcs_actions/src/connection.rs | 2 + .../vcs_actions/src/connection/action_service.rs | 129 +++++++++++++++++++++ crates/vcs_actions/src/connection/error.rs | 14 +++ crates/vcs_data/todo.md | 31 +++++ 4 files changed, 176 insertions(+) create mode 100644 crates/vcs_actions/src/connection.rs create mode 100644 crates/vcs_actions/src/connection/action_service.rs create mode 100644 crates/vcs_actions/src/connection/error.rs create mode 100644 crates/vcs_data/todo.md diff --git a/crates/vcs_actions/src/connection.rs b/crates/vcs_actions/src/connection.rs new file mode 100644 index 0000000..dabbd44 --- /dev/null +++ b/crates/vcs_actions/src/connection.rs @@ -0,0 +1,2 @@ +pub mod action_service; +pub mod error; 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()) + } +} diff --git a/crates/vcs_data/todo.md b/crates/vcs_data/todo.md new file mode 100644 index 0000000..3c7e0c0 --- /dev/null +++ b/crates/vcs_data/todo.md @@ -0,0 +1,31 @@ +| 类别 | 项 | 可完成性 | 已完成 | +|----------|----|----------|--------| +| 本地文件 | 设置上游服务器（仅设置，不会连接和修改染色标识） | y | | +| 本地文件 | 验证连接、权限，并为当前工作区染色（若已染色，则无法连接不同标识的服务器） | y | | +| 本地文件 | 进入表 （否则无法做任何操作） | | | +| 本地文件 | 退出表 （文件将会从当前目录移出，等待下次进入时还原） | | | +| 本地文件 | 去色 - 断开与上游服务器的关联 | y | | +| 本地文件 | 跟踪本地文件的移动、重命名，立刻同步至表 | | | +| 本地文件 | 扫描本地文件结构，标记变化 | | | +| 本地文件 | 通过本地暂存的表索引搜索文件 | | | +| 本地文件 | 查询本地某个文件的状态 | | | +| 本地文件 | 查询当前目录的状态 | | | +| 本地文件 | 查询工作区状态 | | | +| 本地文件 | 将本地所有文件更新到最新状态 | | | +| 本地文件 | 提交所有产生变化的自身所属文件 | | | +| 表 | 表查看 - 指定表并查看结构 | | | +| 表 | 从参照表拉入文件项目 | | | +| 表 | 将文件项目（或多个）导出到指定表 | | | +| 表 | 查看导入请求 | | | +| 表 | 在某个本地地址同意并导入文件 | | | +| 表 | 拒绝某个、某些或所有导入请求 | | | +| 表 | 删除表中的映射，但要确保实际文件已被移除 (忽略文件) | | | +| 表 | 放弃表，所有者消失，下一个切换至表的人获得（放弃需要确保表中没有任何文件是所有者持有的）（替代目前的安全删除） | | | +| 虚拟文件 | 跟踪本地某些文件，并将其创建为虚拟文件，然后添加到自己的表 | | | +| 虚拟文件 | 根据本地文件的目录查找虚拟文件，并为自己获得所有权（需要确保版本和上游同步才可） | | | +| 虚拟文件 | 根据本地文件的目录查找虚拟文件，并放弃所有权（需要确保和上游同步才可） | | | +| 虚拟文件 | 根据本地文件的目录查找虚拟文件，并定向到指定的存在的老版本 | | | + + +？为什么虚拟文件不能删除：虚拟文件的唯一删除方式就是，没有人再用他 +？为什么没有删除表：同理，表权限可以转移，但是删除只能等待定期清除无主人的表 -- cgit From 860fb317bca61ce66a2c98df933aa666dae0a43f Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 13 Oct 2025 11:17:00 +0800 Subject: feat: Add JSON-based action invocation to ActionPool - Extend action_gen macro to generate JSON serialization logic - Implement generic action processing using JSON text for type-agnostic calls - Add callback mechanism with action name and arguments in ActionPool - Update client and server registries to use new callback system - Improve action system flexibility at the cost of serialization overhead --- Cargo.lock | 5 ++++ crates/system_action/Cargo.toml | 1 + crates/system_action/action_macros/Cargo.toml | 4 +++ crates/system_action/action_macros/src/lib.rs | 34 ++++++++++++---------- crates/system_action/src/action_pool.rs | 4 ++- crates/vcs_actions/src/registry/client_registry.rs | 4 +-- crates/vcs_actions/src/registry/server_registry.rs | 4 +-- examples/Cargo.toml | 4 +++ examples/src/bin/example_action_system.rs | 4 +-- 9 files changed, 42 insertions(+), 22 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 247815e..b7911b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8,6 +8,7 @@ version = "0.1.0" dependencies = [ "action_system_macros", "serde", + "serde_json", "tcp_connection", "tokio", ] @@ -18,6 +19,8 @@ version = "0.1.0" dependencies = [ "proc-macro2", "quote", + "serde", + "serde_json", "string_proc", "syn", "tcp_connection", @@ -419,6 +422,8 @@ version = "0.1.0" dependencies = [ "action_system", "cfg_file", + "serde", + "serde_json", "string_proc", "tcp_connection", "tokio", diff --git a/crates/system_action/Cargo.toml b/crates/system_action/Cargo.toml index 120cb34..54ae454 100644 --- a/crates/system_action/Cargo.toml +++ b/crates/system_action/Cargo.toml @@ -9,6 +9,7 @@ action_system_macros = { path = "action_macros" } # Serialization serde = { version = "1.0.219", features = ["derive"] } +serde_json = "1.0.140" # Async & Networking tokio = { version = "1.46.1", features = ["full"] } diff --git a/crates/system_action/action_macros/Cargo.toml b/crates/system_action/action_macros/Cargo.toml index 5ae14fa..869dcde 100644 --- a/crates/system_action/action_macros/Cargo.toml +++ b/crates/system_action/action_macros/Cargo.toml @@ -13,3 +13,7 @@ string_proc = { path = "../../utils/string_proc" } syn = { version = "2.0", features = ["full", "extra-traits"] } quote = "1.0" proc-macro2 = "1.0" + +# Serialization +serde = { version = "1.0.219", features = ["derive"] } +serde_json = "1.0.140" diff --git a/crates/system_action/action_macros/src/lib.rs b/crates/system_action/action_macros/src/lib.rs index 04e974a..683efcb 100644 --- a/crates/system_action/action_macros/src/lib.rs +++ b/crates/system_action/action_macros/src/lib.rs @@ -37,6 +37,9 @@ fn generate_action_struct(input_fn: ItemFn, _is_local: bool) -> proc_macro2::Tok let action_name_ident = &fn_name; + let register_this_action = quote::format_ident!("register_{}", action_name_ident); + let proc_this_action = quote::format_ident!("proc_{}", action_name_ident); + quote! { #[derive(Debug, Clone, Default)] #fn_vis struct #struct_name; @@ -55,22 +58,23 @@ fn generate_action_struct(input_fn: ItemFn, _is_local: bool) -> proc_macro2::Tok } } - impl #struct_name { - #fn_vis fn register_to_pool(pool: &mut action_system::action_pool::ActionPool) { - pool.register::<#struct_name, #arg_type, #return_type>(); - } + #fn_vis fn #register_this_action(pool: &mut action_system::action_pool::ActionPool) { + pool.register::<#struct_name, #arg_type, #return_type>(); + } - #fn_vis async fn process_at_pool<'a>( - pool: &'a action_system::action_pool::ActionPool, - ctx: action_system::action::ActionContext, - #arg_param_name: #arg_type - ) -> Result<#return_type, tcp_connection::error::TcpTargetError> { - pool.process::<#arg_type, #return_type>( - Box::leak(string_proc::snake_case!(stringify!(#action_name_ident)).into_boxed_str()), - ctx, - #arg_param_name - ).await - } + #fn_vis async fn #proc_this_action( + pool: &action_system::action_pool::ActionPool, + ctx: action_system::action::ActionContext, + #arg_param_name: #arg_type + ) -> Result<#return_type, tcp_connection::error::TcpTargetError> { + pool.process::<#arg_type, #return_type>( + Box::leak(string_proc::snake_case!(stringify!(#action_name_ident)).into_boxed_str()), + ctx, + serde_json::to_string(&#arg_param_name) + .map_err(|e| { + tcp_connection::error::TcpTargetError::Serialization(e.to_string()) + })? + ).await } #[allow(dead_code)] diff --git a/crates/system_action/src/action_pool.rs b/crates/system_action/src/action_pool.rs index bc60f7f..a3e82d6 100644 --- a/crates/system_action/src/action_pool.rs +++ b/crates/system_action/src/action_pool.rs @@ -72,7 +72,7 @@ impl ActionPool { &'a self, action_name: &'a str, context: ActionContext, - args: Args, + args_json: String, ) -> Result where Args: serde::de::DeserializeOwned + Send + 'static, @@ -80,6 +80,8 @@ impl ActionPool { { if let Some(action) = self.actions.get(action_name) { let _ = self.exec_on_proc_begin(&context).await?; + let args: Args = serde_json::from_str(&args_json) + .map_err(|e| TcpTargetError::Serialization(format!("Deserialize failed: {}", e)))?; let result = action.process_erased(context, Box::new(args)).await?; let result = *result .downcast::() diff --git a/crates/vcs_actions/src/registry/client_registry.rs b/crates/vcs_actions/src/registry/client_registry.rs index c32ce5a..d298099 100644 --- a/crates/vcs_actions/src/registry/client_registry.rs +++ b/crates/vcs_actions/src/registry/client_registry.rs @@ -1,11 +1,11 @@ use action_system::{action::ActionContext, action_pool::ActionPool}; use tcp_connection::error::TcpTargetError; -use crate::actions::local_actions::SetUpstreamVaultAction; +use crate::actions::local_actions::register_set_upstream_vault_action; fn register_actions(pool: &mut ActionPool) { // Pool register here - SetUpstreamVaultAction::register_to_pool(pool); + register_set_upstream_vault_action(pool); } pub fn client_action_pool() -> ActionPool { diff --git a/crates/vcs_actions/src/registry/server_registry.rs b/crates/vcs_actions/src/registry/server_registry.rs index bdd6a65..3ecc103 100644 --- a/crates/vcs_actions/src/registry/server_registry.rs +++ b/crates/vcs_actions/src/registry/server_registry.rs @@ -1,9 +1,9 @@ use action_system::action_pool::ActionPool; -use crate::actions::local_actions::SetUpstreamVaultAction; +use crate::actions::local_actions::register_set_upstream_vault_action; pub fn server_action_pool() -> ActionPool { let mut pool = ActionPool::new(); - SetUpstreamVaultAction::register_to_pool(&mut pool); + register_set_upstream_vault_action(&mut pool); pool } diff --git a/examples/Cargo.toml b/examples/Cargo.toml index 3a6cee2..f2442ba 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -19,3 +19,7 @@ action_system = { path = "../crates/system_action" } # Async & Networking tokio = { version = "1.46.1", features = ["full"] } + +# Serialization +serde = { version = "1.0.219", features = ["derive"] } +serde_json = "1.0.140" diff --git a/examples/src/bin/example_action_system.rs b/examples/src/bin/example_action_system.rs index a659eb3..c873a2e 100644 --- a/examples/src/bin/example_action_system.rs +++ b/examples/src/bin/example_action_system.rs @@ -4,9 +4,9 @@ use tcp_connection::error::TcpTargetError; #[tokio::main] async fn main() { let mut pool = ActionPool::new(); - PrintNameAction::register_to_pool(&mut pool); + register_print_name_action(&mut pool); - PrintNameAction::process_at_pool(&pool, ActionContext::local(), "World".to_string()) + proc_print_name_action(&pool, ActionContext::local(), "World".to_string()) .await .unwrap(); } -- 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 --- crates/system_action/action_macros/src/lib.rs | 17 ++++-- crates/system_action/src/action.rs | 38 ++++++++++++ crates/system_action/src/action_pool.rs | 67 ++++++++++++++++++++-- crates/vcs_actions/src/actions/local_actions.rs | 2 +- crates/vcs_actions/src/connection.rs | 1 + .../vcs_actions/src/connection/action_service.rs | 29 ++++++++-- crates/vcs_actions/src/connection/protocol.rs | 7 +++ crates/vcs_actions/src/registry/client_registry.rs | 28 +++++++-- 8 files changed, 166 insertions(+), 23 deletions(-) create mode 100644 crates/vcs_actions/src/connection/protocol.rs diff --git a/crates/system_action/action_macros/src/lib.rs b/crates/system_action/action_macros/src/lib.rs index 683efcb..aa1c696 100644 --- a/crates/system_action/action_macros/src/lib.rs +++ b/crates/system_action/action_macros/src/lib.rs @@ -67,14 +67,19 @@ fn generate_action_struct(input_fn: ItemFn, _is_local: bool) -> proc_macro2::Tok ctx: action_system::action::ActionContext, #arg_param_name: #arg_type ) -> Result<#return_type, tcp_connection::error::TcpTargetError> { - pool.process::<#arg_type, #return_type>( + let args_json = serde_json::to_string(&#arg_param_name) + .map_err(|e| { + tcp_connection::error::TcpTargetError::Serialization(e.to_string()) + })?; + let result_json = pool.process_json( Box::leak(string_proc::snake_case!(stringify!(#action_name_ident)).into_boxed_str()), ctx, - serde_json::to_string(&#arg_param_name) - .map_err(|e| { - tcp_connection::error::TcpTargetError::Serialization(e.to_string()) - })? - ).await + args_json, + ).await?; + serde_json::from_str(&result_json) + .map_err(|e| { + tcp_connection::error::TcpTargetError::Serialization(e.to_string()) + }) } #[allow(dead_code)] diff --git a/crates/system_action/src/action.rs b/crates/system_action/src/action.rs index c8d7a18..e7d2d8c 100644 --- a/crates/system_action/src/action.rs +++ b/crates/system_action/src/action.rs @@ -22,6 +22,12 @@ pub struct ActionContext { /// Whether the action is executed locally or remotely local: bool, + /// The name of the action being executed + action_name: String, + + /// The JSON-serialized arguments for the action + action_args_json: String, + /// The connection instance in the current context, /// used to interact with the machine on the other end instance: Option, @@ -53,6 +59,11 @@ impl ActionContext { self.instance = Some(instance); self } + + /// Pop connection instance from context + pub fn pop_instance(&mut self) -> Option { + self.instance.take() + } } impl ActionContext { @@ -70,4 +81,31 @@ impl ActionContext { pub fn instance(&self) -> &Option { &self.instance } + + /// Get a mutable reference to the connection instance in the current context + pub fn instance_mut(&mut self) -> &mut Option { + &mut self.instance + } + + /// Get the action name from the context + pub fn action_name(&self) -> &str { + &self.action_name + } + + /// Get the action arguments from the context + pub fn action_args_json(&self) -> &String { + &self.action_args_json + } + + /// Set the action name in the context + pub fn set_action_name(mut self, action_name: String) -> Self { + self.action_name = action_name; + self + } + + /// Set the action arguments in the context + pub fn set_action_args_json(mut self, action_args: String) -> Self { + self.action_args_json = action_args; + self + } } diff --git a/crates/system_action/src/action_pool.rs b/crates/system_action/src/action_pool.rs index a3e82d6..7e93fc4 100644 --- a/crates/system_action/src/action_pool.rs +++ b/crates/system_action/src/action_pool.rs @@ -1,13 +1,14 @@ use std::pin::Pin; use serde::{Serialize, de::DeserializeOwned}; +use serde_json; use tcp_connection::error::TcpTargetError; use crate::action::{Action, ActionContext}; type ProcBeginCallback = for<'a> fn( - &'a ActionContext, + &'a mut ActionContext, ) -> Pin> + Send + 'a>>; type ProcEndCallback = fn() -> Pin> + Send>>; @@ -68,20 +69,51 @@ impl ActionPool { /// ```ignore /// let result = action_pool.process::("my_action", context, args).await?; /// ``` - pub async fn process<'a, Args, Return>( + /// Processes an action by name with JSON-serialized arguments + /// + /// Usage: + /// ```ignore + /// let result_json = action_pool.process_json("my_action", context, args_json).await?; + /// let result: MyReturn = serde_json::from_str(&result_json)?; + /// ``` + pub async fn process_json<'a>( &'a self, action_name: &'a str, context: ActionContext, args_json: String, + ) -> Result { + if let Some(action) = self.actions.get(action_name) { + // Set action name and args in context for callbacks + let context = context.set_action_name(action_name.to_string()); + let mut context = context.set_action_args_json(args_json.clone()); + + let _ = self.exec_on_proc_begin(&mut context).await?; + let result = action.process_json_erased(context, args_json).await?; + let _ = self.exec_on_proc_end().await?; + Ok(result) + } else { + Err(TcpTargetError::Unsupported("InvalidAction".to_string())) + } + } + + /// Processes an action by name with given context and arguments + /// + /// Usage: + /// ```ignore + /// let result = action_pool.process::("my_action", context, args).await?; + /// ``` + pub async fn process<'a, Args, Return>( + &'a self, + action_name: &'a str, + mut context: ActionContext, + args: Args, ) -> Result where Args: serde::de::DeserializeOwned + Send + 'static, Return: serde::Serialize + Send + 'static, { if let Some(action) = self.actions.get(action_name) { - let _ = self.exec_on_proc_begin(&context).await?; - let args: Args = serde_json::from_str(&args_json) - .map_err(|e| TcpTargetError::Serialization(format!("Deserialize failed: {}", e)))?; + let _ = self.exec_on_proc_begin(&mut context).await?; let result = action.process_erased(context, Box::new(args)).await?; let result = *result .downcast::() @@ -94,7 +126,7 @@ impl ActionPool { } /// Executes the process begin callback if set - async fn exec_on_proc_begin(&self, context: &ActionContext) -> Result<(), TcpTargetError> { + async fn exec_on_proc_begin(&self, context: &mut ActionContext) -> Result<(), TcpTargetError> { if let Some(callback) = &self.on_proc_begin { callback(context).await } else { @@ -125,6 +157,13 @@ trait ActionErased: Send + Sync { + Send, >, >; + + /// Processes the action with JSON-serialized arguments and returns JSON-serialized result + fn process_json_erased( + &self, + context: ActionContext, + args_json: String, + ) -> std::pin::Pin> + Send>>; } /// Wrapper struct that implements ActionErased for concrete Action types @@ -154,4 +193,20 @@ where Ok(Box::new(result) as Box) }) } + + fn process_json_erased( + &self, + context: ActionContext, + args_json: String, + ) -> std::pin::Pin> + Send>> + { + Box::pin(async move { + let args: Args = serde_json::from_str(&args_json) + .map_err(|e| TcpTargetError::Serialization(format!("Deserialize failed: {}", e)))?; + let result = A::process(context, args).await?; + let result_json = serde_json::to_string(&result) + .map_err(|e| TcpTargetError::Serialization(format!("Serialize failed: {}", e)))?; + Ok(result_json) + }) + } } diff --git a/crates/vcs_actions/src/actions/local_actions.rs b/crates/vcs_actions/src/actions/local_actions.rs index b230c6f..0e210a7 100644 --- a/crates/vcs_actions/src/actions/local_actions.rs +++ b/crates/vcs_actions/src/actions/local_actions.rs @@ -6,7 +6,7 @@ use tcp_connection::error::TcpTargetError; #[action_gen(local)] pub async fn set_upstream_vault_action( ctx: ActionContext, - upstream: SocketAddr, + _upstream: SocketAddr, ) -> Result<(), TcpTargetError> { if ctx.is_remote() { return Err(TcpTargetError::NotLocal( diff --git a/crates/vcs_actions/src/connection.rs b/crates/vcs_actions/src/connection.rs index dabbd44..918f93c 100644 --- a/crates/vcs_actions/src/connection.rs +++ b/crates/vcs_actions/src/connection.rs @@ -1,2 +1,3 @@ pub mod action_service; pub mod error; +pub mod protocol; 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, +} diff --git a/crates/vcs_actions/src/registry/client_registry.rs b/crates/vcs_actions/src/registry/client_registry.rs index d298099..56acdad 100644 --- a/crates/vcs_actions/src/registry/client_registry.rs +++ b/crates/vcs_actions/src/registry/client_registry.rs @@ -1,7 +1,10 @@ use action_system::{action::ActionContext, action_pool::ActionPool}; use tcp_connection::error::TcpTargetError; -use crate::actions::local_actions::register_set_upstream_vault_action; +use crate::{ + actions::local_actions::register_set_upstream_vault_action, + connection::protocol::RemoteActionInvoke, +}; fn register_actions(pool: &mut ActionPool) { // Pool register here @@ -22,18 +25,33 @@ pub fn client_action_pool() -> ActionPool { pool } -async fn on_proc_begin(ctx: &ActionContext) -> Result<(), TcpTargetError> { +async fn on_proc_begin(ctx: &mut ActionContext) -> Result<(), TcpTargetError> { + // Is ctx remote + let is_remote = ctx.is_remote(); + + // Action name and arguments + let action_name = ctx.action_name().to_string(); + let action_args_json = ctx.action_args_json().clone(); + // Get instance - let Some(_instance) = ctx.instance() else { + let Some(instance) = ctx.instance_mut() else { return Err(TcpTargetError::Unsupported( "Missing ConnectionInstance in current context, this ActionPool does not support this call" .to_string())); }; // If it's remote, invoke action at server - if ctx.is_remote() { - // instance.write_text(text) + if is_remote { + // Build protocol message + let msg = RemoteActionInvoke { + action_name: action_name, + action_args_json: action_args_json, + }; + + // Send + instance.write_msgpack(msg).await?; } + // Return OK, wait for client to execute Action locally Ok(()) } -- 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/system_action/action_macros/src/lib.rs | 4 +- crates/system_action/src/action.rs | 70 +++++++++++++++++++--- crates/system_action/src/action_pool.rs | 19 +++--- crates/vcs_actions/src/actions/local_actions.rs | 28 +++++---- .../vcs_actions/src/connection/action_service.rs | 5 +- crates/vcs_actions/src/registry/client_registry.rs | 17 +++--- crates/vcs_actions/src/registry/server_registry.rs | 4 +- 7 files changed, 109 insertions(+), 38 deletions(-) diff --git a/crates/system_action/action_macros/src/lib.rs b/crates/system_action/action_macros/src/lib.rs index aa1c696..d1a47ee 100644 --- a/crates/system_action/action_macros/src/lib.rs +++ b/crates/system_action/action_macros/src/lib.rs @@ -89,12 +89,12 @@ fn generate_action_struct(input_fn: ItemFn, _is_local: bool) -> proc_macro2::Tok #[doc = ""] #[doc = "Register the action to the pool."] #[doc = "```ignore"] - #[doc = "YourAction::register_to_pool(&mut pool);"] + #[doc = "register_your_func(&mut pool);"] #[doc = "```"] #[doc = ""] #[doc = "Process the action at the pool."] #[doc = "```ignore"] - #[doc = "let result = YourAction::process_at_pool(&pool, ctx, arg).await?;"] + #[doc = "let result = proc_your_func(&pool, ctx, arg).await?;"] #[doc = "```"] #fn_vis #fn_sig #fn_block } diff --git a/crates/system_action/src/action.rs b/crates/system_action/src/action.rs index e7d2d8c..3ae5711 100644 --- a/crates/system_action/src/action.rs +++ b/crates/system_action/src/action.rs @@ -1,6 +1,9 @@ use serde::{Serialize, de::DeserializeOwned}; +use std::any::{Any, TypeId}; +use std::collections::HashMap; +use std::sync::Arc; use tcp_connection::{error::TcpTargetError, instance::ConnectionInstance}; -use tokio::net::TcpStream; +use tokio::{net::TcpStream, sync::Mutex}; pub trait Action where @@ -29,8 +32,10 @@ pub struct ActionContext { action_args_json: String, /// The connection instance in the current context, - /// used to interact with the machine on the other end - instance: Option, + instance: Option>>, + + /// Generic data storage for arbitrary types + data: HashMap>, } impl ActionContext { @@ -50,18 +55,18 @@ impl ActionContext { /// Build connection instance from TcpStream pub fn build_instance(mut self, stream: TcpStream) -> Self { - self.instance = Some(ConnectionInstance::from(stream)); + self.instance = Some(Arc::new(Mutex::new(ConnectionInstance::from(stream)))); self } /// Insert connection instance into context pub fn insert_instance(mut self, instance: ConnectionInstance) -> Self { - self.instance = Some(instance); + self.instance = Some(Arc::new(Mutex::new(instance))); self } /// Pop connection instance from context - pub fn pop_instance(&mut self) -> Option { + pub fn pop_instance(&mut self) -> Option>> { self.instance.take() } } @@ -78,12 +83,12 @@ impl ActionContext { } /// Get the connection instance in the current context - pub fn instance(&self) -> &Option { + pub fn instance(&self) -> &Option>> { &self.instance } /// Get a mutable reference to the connection instance in the current context - pub fn instance_mut(&mut self) -> &mut Option { + pub fn instance_mut(&mut self) -> &mut Option>> { &mut self.instance } @@ -104,8 +109,55 @@ impl ActionContext { } /// Set the action arguments in the context - pub fn set_action_args_json(mut self, action_args: String) -> Self { + pub fn set_action_args(mut self, action_args: String) -> Self { self.action_args_json = action_args; self } + + /// Insert arbitrary data in the context + pub fn insert(mut self, value: T) -> Self { + self.data.insert(TypeId::of::(), Arc::new(value)); + self + } + + /// Insert arbitrary data as Arc in the context + pub fn insert_arc(mut self, value: Arc) -> Self { + self.data.insert(TypeId::of::(), value); + self + } + + /// Get arbitrary data from the context + pub fn get(&self) -> Option<&T> { + self.data + .get(&TypeId::of::()) + .and_then(|arc| arc.downcast_ref::()) + } + + /// Get arbitrary data as Arc from the context + pub fn get_arc(&self) -> Option> { + self.data + .get(&TypeId::of::()) + .and_then(|arc| Arc::clone(arc).downcast::().ok()) + } + + /// Remove and return arbitrary data from the context + pub fn remove(&mut self) -> Option> { + self.data + .remove(&TypeId::of::()) + .and_then(|arc| arc.downcast::().ok()) + } + + /// Check if the context contains data of a specific type + pub fn contains(&self) -> bool { + self.data.contains_key(&TypeId::of::()) + } + + /// Take ownership of the context and extract data of a specific type + pub fn take(mut self) -> (Self, Option>) { + let value = self + .data + .remove(&TypeId::of::()) + .and_then(|arc| arc.downcast::().ok()); + (self, value) + } } diff --git a/crates/system_action/src/action_pool.rs b/crates/system_action/src/action_pool.rs index 7e93fc4..f3e178a 100644 --- a/crates/system_action/src/action_pool.rs +++ b/crates/system_action/src/action_pool.rs @@ -8,7 +8,8 @@ use crate::action::{Action, ActionContext}; type ProcBeginCallback = for<'a> fn( - &'a mut ActionContext, + &'a ActionContext, + args: &'a (dyn std::any::Any + Send + Sync), ) -> Pin> + Send + 'a>>; type ProcEndCallback = fn() -> Pin> + Send>>; @@ -85,9 +86,9 @@ impl ActionPool { if let Some(action) = self.actions.get(action_name) { // Set action name and args in context for callbacks let context = context.set_action_name(action_name.to_string()); - let mut context = context.set_action_args_json(args_json.clone()); + let context = context.set_action_args(args_json.clone()); - let _ = self.exec_on_proc_begin(&mut context).await?; + let _ = self.exec_on_proc_begin(&context, &args_json).await?; let result = action.process_json_erased(context, args_json).await?; let _ = self.exec_on_proc_end().await?; Ok(result) @@ -109,11 +110,11 @@ impl ActionPool { args: Args, ) -> Result where - Args: serde::de::DeserializeOwned + Send + 'static, + Args: serde::de::DeserializeOwned + Send + Sync + 'static, Return: serde::Serialize + Send + 'static, { if let Some(action) = self.actions.get(action_name) { - let _ = self.exec_on_proc_begin(&mut context).await?; + let _ = self.exec_on_proc_begin(&context, &args).await?; let result = action.process_erased(context, Box::new(args)).await?; let result = *result .downcast::() @@ -126,9 +127,13 @@ impl ActionPool { } /// Executes the process begin callback if set - async fn exec_on_proc_begin(&self, context: &mut ActionContext) -> Result<(), TcpTargetError> { + async fn exec_on_proc_begin( + &self, + context: &ActionContext, + args: &(dyn std::any::Any + Send + Sync), + ) -> Result<(), TcpTargetError> { if let Some(callback) = &self.on_proc_begin { - callback(context).await + callback(context, args).await } else { Ok(()) } diff --git a/crates/vcs_actions/src/actions/local_actions.rs b/crates/vcs_actions/src/actions/local_actions.rs index 0e210a7..b11a934 100644 --- a/crates/vcs_actions/src/actions/local_actions.rs +++ b/crates/vcs_actions/src/actions/local_actions.rs @@ -1,17 +1,25 @@ -use std::net::SocketAddr; - use action_system::{action::ActionContext, action_gen}; +use log::info; use tcp_connection::error::TcpTargetError; -#[action_gen(local)] -pub async fn set_upstream_vault_action( - ctx: ActionContext, - _upstream: SocketAddr, -) -> Result<(), TcpTargetError> { - if ctx.is_remote() { - return Err(TcpTargetError::NotLocal( - "Action was not invoked on the local machine".to_string(), +#[action_gen] +pub async fn hello_world_action(ctx: ActionContext, _n: ()) -> Result<(), TcpTargetError> { + // Ensure the instance is available + let Some(instance) = ctx.instance() else { + return Err(TcpTargetError::NotFound( + "Connection Instance Lost.".to_string(), )); + }; + + if ctx.is_local() { + // Invoke on local + // Send the message to the server + let _ = instance.lock().await.write_text("Hello World!").await; + } else if ctx.is_remote() { + // Read the message from the client + let read = instance.lock().await.read_text().await?; + info!("{}", read) } + Ok(()) } 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 diff --git a/crates/vcs_actions/src/registry/client_registry.rs b/crates/vcs_actions/src/registry/client_registry.rs index 56acdad..47fd7ee 100644 --- a/crates/vcs_actions/src/registry/client_registry.rs +++ b/crates/vcs_actions/src/registry/client_registry.rs @@ -2,13 +2,12 @@ use action_system::{action::ActionContext, action_pool::ActionPool}; use tcp_connection::error::TcpTargetError; use crate::{ - actions::local_actions::register_set_upstream_vault_action, - connection::protocol::RemoteActionInvoke, + actions::local_actions::register_hello_world_action, connection::protocol::RemoteActionInvoke, }; fn register_actions(pool: &mut ActionPool) { // Pool register here - register_set_upstream_vault_action(pool); + register_hello_world_action(pool); } pub fn client_action_pool() -> ActionPool { @@ -19,13 +18,16 @@ pub fn client_action_pool() -> ActionPool { register_actions(&mut pool); // Add process events - pool.set_on_proc_begin(|ctx| Box::pin(on_proc_begin(ctx))); + pool.set_on_proc_begin(|ctx, args| Box::pin(on_proc_begin(ctx, args))); // Return pool } -async fn on_proc_begin(ctx: &mut ActionContext) -> Result<(), TcpTargetError> { +async fn on_proc_begin( + ctx: &ActionContext, + _args: &(dyn std::any::Any + Send + Sync), +) -> Result<(), TcpTargetError> { // Is ctx remote let is_remote = ctx.is_remote(); @@ -34,7 +36,7 @@ async fn on_proc_begin(ctx: &mut ActionContext) -> Result<(), TcpTargetError> { let action_args_json = ctx.action_args_json().clone(); // Get instance - let Some(instance) = ctx.instance_mut() else { + let Some(instance) = ctx.instance() else { return Err(TcpTargetError::Unsupported( "Missing ConnectionInstance in current context, this ActionPool does not support this call" .to_string())); @@ -49,7 +51,8 @@ async fn on_proc_begin(ctx: &mut ActionContext) -> Result<(), TcpTargetError> { }; // Send - instance.write_msgpack(msg).await?; + let mut instance = instance.lock().await; + instance.write_msgpack(&msg).await?; } // Return OK, wait for client to execute Action locally diff --git a/crates/vcs_actions/src/registry/server_registry.rs b/crates/vcs_actions/src/registry/server_registry.rs index 3ecc103..b449b68 100644 --- a/crates/vcs_actions/src/registry/server_registry.rs +++ b/crates/vcs_actions/src/registry/server_registry.rs @@ -1,9 +1,9 @@ use action_system::action_pool::ActionPool; -use crate::actions::local_actions::register_set_upstream_vault_action; +use crate::actions::local_actions::register_hello_world_action; pub fn server_action_pool() -> ActionPool { let mut pool = ActionPool::new(); - register_set_upstream_vault_action(&mut pool); + register_hello_world_action(&mut pool); 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/system_action/action_macros/src/lib.rs | 7 ++-- crates/system_action/src/action.rs | 14 ++++--- crates/system_action/src/action_pool.rs | 48 ++++++++++++++-------- .../tcp_connection_test/src/test_msgpack.rs | 11 +---- .../vcs_actions/src/connection/action_service.rs | 4 +- crates/vcs_actions/src/registry/client_registry.rs | 4 +- crates/vcs_data/src/data/sheet.rs | 8 ++-- ...st_sheet_creation_management_and_persistence.rs | 2 +- 8 files changed, 51 insertions(+), 47 deletions(-) diff --git a/crates/system_action/action_macros/src/lib.rs b/crates/system_action/action_macros/src/lib.rs index d1a47ee..ce50073 100644 --- a/crates/system_action/action_macros/src/lib.rs +++ b/crates/system_action/action_macros/src/lib.rs @@ -101,7 +101,7 @@ fn generate_action_struct(input_fn: ItemFn, _is_local: bool) -> proc_macro2::Tok } fn validate_function_signature(fn_sig: &syn::Signature) { - if !fn_sig.asyncness.is_some() { + if fn_sig.asyncness.is_none() { panic!("Expected async function for Action, but found synchronous function"); } @@ -120,13 +120,12 @@ fn validate_function_signature(fn_sig: &syn::Signature) { }; if let syn::Type::Path(type_path) = return_type.as_ref() { - if let Some(segment) = type_path.path.segments.last() { - if segment.ident != "Result" { + if let Some(segment) = type_path.path.segments.last() + && segment.ident != "Result" { panic!( "Expected Action function to return Result, but found different return type" ); } - } } else { panic!( "Expected Action function to return Result, but found no return type" diff --git a/crates/system_action/src/action.rs b/crates/system_action/src/action.rs index 3ae5711..8a6180a 100644 --- a/crates/system_action/src/action.rs +++ b/crates/system_action/src/action.rs @@ -41,16 +41,18 @@ pub struct ActionContext { impl ActionContext { /// Generate local context pub fn local() -> Self { - let mut ctx = ActionContext::default(); - ctx.local = true; - ctx + ActionContext { + local: true, + ..Default::default() + } } /// Generate remote context pub fn remote() -> Self { - let mut ctx = ActionContext::default(); - ctx.local = false; - ctx + ActionContext { + local: false, + ..Default::default() + } } /// Build connection instance from TcpStream diff --git a/crates/system_action/src/action_pool.rs b/crates/system_action/src/action_pool.rs index f3e178a..c28de1e 100644 --- a/crates/system_action/src/action_pool.rs +++ b/crates/system_action/src/action_pool.rs @@ -6,12 +6,14 @@ use tcp_connection::error::TcpTargetError; use crate::action::{Action, ActionContext}; -type ProcBeginCallback = - for<'a> fn( - &'a ActionContext, - args: &'a (dyn std::any::Any + Send + Sync), - ) -> Pin> + Send + 'a>>; -type ProcEndCallback = fn() -> Pin> + Send>>; +type ProcBeginCallback = for<'a> fn( + &'a ActionContext, + args: &'a (dyn std::any::Any + Send + Sync), +) -> ProcBeginFuture<'a>; +type ProcEndCallback = fn() -> ProcEndFuture; + +type ProcBeginFuture<'a> = Pin> + Send + 'a>>; +type ProcEndFuture = Pin> + Send>>; /// A pool of registered actions that can be processed by name pub struct ActionPool { @@ -25,6 +27,12 @@ pub struct ActionPool { on_proc_end: Option, } +impl Default for ActionPool { + fn default() -> Self { + Self::new() + } +} + impl ActionPool { /// Creates a new empty ActionPool pub fn new() -> Self { @@ -88,9 +96,9 @@ impl ActionPool { let context = context.set_action_name(action_name.to_string()); let context = context.set_action_args(args_json.clone()); - let _ = self.exec_on_proc_begin(&context, &args_json).await?; + self.exec_on_proc_begin(&context, &args_json).await?; let result = action.process_json_erased(context, args_json).await?; - let _ = self.exec_on_proc_end().await?; + self.exec_on_proc_end().await?; Ok(result) } else { Err(TcpTargetError::Unsupported("InvalidAction".to_string())) @@ -106,7 +114,7 @@ impl ActionPool { pub async fn process<'a, Args, Return>( &'a self, action_name: &'a str, - mut context: ActionContext, + context: ActionContext, args: Args, ) -> Result where @@ -114,12 +122,12 @@ impl ActionPool { Return: serde::Serialize + Send + 'static, { if let Some(action) = self.actions.get(action_name) { - let _ = self.exec_on_proc_begin(&context, &args).await?; + self.exec_on_proc_begin(&context, &args).await?; let result = action.process_erased(context, Box::new(args)).await?; let result = *result .downcast::() .map_err(|_| TcpTargetError::Unsupported("InvalidArguments".to_string()))?; - let _ = self.exec_on_proc_end().await?; + self.exec_on_proc_end().await?; Ok(result) } else { Err(TcpTargetError::Unsupported("InvalidAction".to_string())) @@ -150,25 +158,29 @@ impl ActionPool { } /// Trait for type-erased actions that can be stored in ActionPool +type ProcessErasedFuture = std::pin::Pin< + Box< + dyn std::future::Future, TcpTargetError>> + + Send, + >, +>; +type ProcessJsonErasedFuture = + std::pin::Pin> + Send>>; + trait ActionErased: Send + Sync { /// Processes the action with type-erased arguments and returns type-erased result fn process_erased( &self, context: ActionContext, args: Box, - ) -> std::pin::Pin< - Box< - dyn std::future::Future, TcpTargetError>> - + Send, - >, - >; + ) -> ProcessErasedFuture; /// Processes the action with JSON-serialized arguments and returns JSON-serialized result fn process_json_erased( &self, context: ActionContext, args_json: String, - ) -> std::pin::Pin> + Send>>; + ) -> ProcessJsonErasedFuture; } /// Wrapper struct that implements ActionErased for concrete Action types diff --git a/crates/utils/tcp_connection/tcp_connection_test/src/test_msgpack.rs b/crates/utils/tcp_connection/tcp_connection_test/src/test_msgpack.rs index 7a7dc1f..4c9c870 100644 --- a/crates/utils/tcp_connection/tcp_connection_test/src/test_msgpack.rs +++ b/crates/utils/tcp_connection/tcp_connection_test/src/test_msgpack.rs @@ -9,21 +9,12 @@ use crate::test_utils::{ target_configure::ServerTargetConfig, }; -#[derive(Debug, PartialEq, Serialize, Deserialize)] +#[derive(Debug, PartialEq, Serialize, Deserialize, Default)] struct TestData { id: u32, name: String, } -impl Default for TestData { - fn default() -> Self { - Self { - id: 0, - name: String::new(), - } - } -} - pub(crate) struct MsgPackClientHandle; impl ClientHandle for MsgPackClientHandle { 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) diff --git a/crates/vcs_actions/src/registry/client_registry.rs b/crates/vcs_actions/src/registry/client_registry.rs index 47fd7ee..5939bed 100644 --- a/crates/vcs_actions/src/registry/client_registry.rs +++ b/crates/vcs_actions/src/registry/client_registry.rs @@ -46,8 +46,8 @@ async fn on_proc_begin( if is_remote { // Build protocol message let msg = RemoteActionInvoke { - action_name: action_name, - action_args_json: action_args_json, + action_name, + action_args_json, }; // Send diff --git a/crates/vcs_data/src/data/sheet.rs b/crates/vcs_data/src/data/sheet.rs index a6220c9..f1cf67c 100644 --- a/crates/vcs_data/src/data/sheet.rs +++ b/crates/vcs_data/src/data/sheet.rs @@ -107,7 +107,7 @@ impl<'a> Sheet<'a> { } /// Accept an input package and insert to the sheet - pub fn accept_import( + pub async fn accept_import( &mut self, input_name: &InputName, insert_to: &SheetPathBuf, @@ -129,7 +129,8 @@ impl<'a> Sheet<'a> { // Insert to sheet for (relative_path, virtual_file_id) in input.files { - let _ = self.add_mapping(insert_to.join(relative_path), virtual_file_id); + self.add_mapping(insert_to.join(relative_path), virtual_file_id) + .await?; } Ok(()) @@ -176,8 +177,7 @@ impl<'a> Sheet<'a> { } Err(_) => { // Error checking rights, don't allow modifying the mapping - Err(std::io::Error::new( - std::io::ErrorKind::Other, + Err(std::io::Error::other( "Failed to check virtual file edit rights", )) } diff --git a/crates/vcs_data/vcs_data_test/src/test_sheet_creation_management_and_persistence.rs b/crates/vcs_data/vcs_data_test/src/test_sheet_creation_management_and_persistence.rs index 461d465..a8dfb89 100644 --- a/crates/vcs_data/vcs_data_test/src/test_sheet_creation_management_and_persistence.rs +++ b/crates/vcs_data/vcs_data_test/src/test_sheet_creation_management_and_persistence.rs @@ -254,7 +254,7 @@ async fn test_sheet_data_serialization() -> Result<(), std::io::Error> { // Add some inputs let input_name = "source_files".to_string(); - let _files = vec![ + let _files = [ ( InputRelativePathBuf::from("src/main.rs"), VirtualFileId::new(), -- cgit