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(-) (limited to 'crates/system_action/src/action_pool.rs') 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 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 --- crates/system_action/src/action_pool.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) (limited to 'crates/system_action/src/action_pool.rs') 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::() -- 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/src/action_pool.rs | 67 ++++++++++++++++++++++++++++++--- 1 file changed, 61 insertions(+), 6 deletions(-) (limited to 'crates/system_action/src/action_pool.rs') 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) + }) + } } -- 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/src/action_pool.rs | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) (limited to 'crates/system_action/src/action_pool.rs') 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(()) } -- 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/src/action_pool.rs | 48 ++++++++++++++++++++------------- 1 file changed, 30 insertions(+), 18 deletions(-) (limited to 'crates/system_action/src/action_pool.rs') 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 -- cgit