summaryrefslogtreecommitdiff
path: root/crates/system_action/src/action_pool.rs
diff options
context:
space:
mode:
author魏曹先生 <1992414357@qq.com>2025-10-13 13:34:39 +0800
committer魏曹先生 <1992414357@qq.com>2025-10-13 13:34:39 +0800
commit67fb8ec01b351c6c9fd2af321166bb92250b1218 (patch)
tree4509509ef932e1009dcd2da25928fcff71a5f29b /crates/system_action/src/action_pool.rs
parent860fb317bca61ce66a2c98df933aa666dae0a43f (diff)
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
Diffstat (limited to 'crates/system_action/src/action_pool.rs')
-rw-r--r--crates/system_action/src/action_pool.rs67
1 files changed, 61 insertions, 6 deletions
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<Box<dyn Future<Output = Result<(), TcpTargetError>> + Send + 'a>>;
type ProcEndCallback = fn() -> Pin<Box<dyn Future<Output = Result<(), TcpTargetError>> + Send>>;
@@ -68,20 +69,51 @@ impl ActionPool {
/// ```ignore
/// let result = action_pool.process::<MyArgs, MyReturn>("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<String, TcpTargetError> {
+ 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::<MyArgs, MyReturn>("my_action", context, args).await?;
+ /// ```
+ pub async fn process<'a, Args, Return>(
+ &'a self,
+ action_name: &'a str,
+ mut context: ActionContext,
+ args: Args,
) -> Result<Return, TcpTargetError>
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::<Return>()
@@ -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<Box<dyn std::future::Future<Output = Result<String, TcpTargetError>> + Send>>;
}
/// Wrapper struct that implements ActionErased for concrete Action types
@@ -154,4 +193,20 @@ where
Ok(Box::new(result) as Box<dyn std::any::Any + Send>)
})
}
+
+ fn process_json_erased(
+ &self,
+ context: ActionContext,
+ args_json: String,
+ ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String, TcpTargetError>> + 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)
+ })
+ }
}