summaryrefslogtreecommitdiff
path: root/systems
diff options
context:
space:
mode:
Diffstat (limited to 'systems')
-rw-r--r--systems/action/Cargo.toml15
-rw-r--r--systems/action/action_macros/Cargo.toml19
-rw-r--r--systems/action/action_macros/src/lib.rs248
-rw-r--r--systems/action/src/action.rs244
-rw-r--r--systems/action/src/action_pool.rs247
-rw-r--r--systems/action/src/lib.rs6
6 files changed, 779 insertions, 0 deletions
diff --git a/systems/action/Cargo.toml b/systems/action/Cargo.toml
new file mode 100644
index 0000000..5317975
--- /dev/null
+++ b/systems/action/Cargo.toml
@@ -0,0 +1,15 @@
+[package]
+name = "action_system"
+edition = "2024"
+version.workspace = true
+
+[dependencies]
+tcp_connection = { path = "../../utils/tcp_connection" }
+action_system_macros = { path = "action_macros" }
+
+# Serialization
+serde = { version = "1.0.228", features = ["derive"] }
+serde_json = "1.0.145"
+
+# Async & Networking
+tokio = "1.48.0"
diff --git a/systems/action/action_macros/Cargo.toml b/systems/action/action_macros/Cargo.toml
new file mode 100644
index 0000000..8b23191
--- /dev/null
+++ b/systems/action/action_macros/Cargo.toml
@@ -0,0 +1,19 @@
+[package]
+name = "action_system_macros"
+edition = "2024"
+version.workspace = true
+
+[lib]
+proc-macro = true
+
+[dependencies]
+tcp_connection = { path = "../../../utils/tcp_connection" }
+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.228", features = ["derive"] }
+serde_json = "1.0.145"
diff --git a/systems/action/action_macros/src/lib.rs b/systems/action/action_macros/src/lib.rs
new file mode 100644
index 0000000..e6616b4
--- /dev/null
+++ b/systems/action/action_macros/src/lib.rs
@@ -0,0 +1,248 @@
+use proc_macro::TokenStream;
+use quote::quote;
+use syn::{ItemFn, parse_macro_input};
+
+/// # Macro - Generate Action
+///
+/// When annotating a function with the `#[action_gen]` macro in the following format, it generates boilerplate code for client-server interaction
+///
+/// ```ignore
+/// #[action_gen]
+/// async fn action_name(ctx: ActionContext, argument: YourArgument) -> Result<YourResult, TcpTargetError> {
+/// // Write your client and server logic here
+/// if ctx.is_proc_on_remote() {
+/// // Server logic
+/// }
+/// if ctx.is_proc_on_local() {
+/// // Client logic
+/// }
+/// }
+/// ```
+///
+/// > WARNING:
+/// > For Argument and Result types, the `action_gen` macro only supports types that derive serde's Serialize and Deserialize
+///
+/// ## Generated Code
+///
+/// `action_gen` will generate the following:
+///
+/// 1. Complete implementation of Action<Args, Return>
+/// 2. Process / Register method
+///
+/// ## How to use
+///
+/// You can use your generated method as follows
+///
+/// ```ignore
+/// async fn main() -> Result<(), TcpTargetError> {
+///
+/// // Prepare your argument
+/// let args = YourArgument::default();
+///
+/// // Create a pool and context
+/// let mut pool = ActionPool::new();
+/// let ctx = ActionContext::local();
+///
+/// // Register your action
+/// register_your_action(&mut pool);
+///
+/// // Process your action
+/// proc_your_action(&pool, ctx, args).await?;
+///
+/// Ok(())
+/// }
+/// ```
+#[proc_macro_attribute]
+pub fn action_gen(attr: TokenStream, item: TokenStream) -> TokenStream {
+ let input_fn = parse_macro_input!(item as ItemFn);
+ let is_local = if attr.is_empty() {
+ false
+ } else {
+ let attr_str = attr.to_string();
+ attr_str == "local" || attr_str.contains("local")
+ };
+
+ generate_action_struct(input_fn, is_local).into()
+}
+
+fn generate_action_struct(input_fn: ItemFn, _is_local: bool) -> proc_macro2::TokenStream {
+ let fn_vis = &input_fn.vis;
+ let fn_sig = &input_fn.sig;
+ let fn_name = &fn_sig.ident;
+ let fn_block = &input_fn.block;
+
+ validate_function_signature(fn_sig);
+
+ let (context_param_name, arg_param_name, arg_type, return_type) =
+ extract_parameters_and_types(fn_sig);
+
+ let struct_name = quote::format_ident!("{}", convert_to_pascal_case(&fn_name.to_string()));
+
+ 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;
+
+ impl action_system::action::Action<#arg_type, #return_type> for #struct_name {
+ fn action_name() -> &'static str {
+ Box::leak(string_proc::snake_case!(stringify!(#action_name_ident)).into_boxed_str())
+ }
+
+ fn is_remote_action() -> bool {
+ !#_is_local
+ }
+
+ async fn process(#context_param_name: action_system::action::ActionContext, #arg_param_name: #arg_type) -> Result<#return_type, tcp_connection::error::TcpTargetError> {
+ #fn_block
+ }
+ }
+
+ #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 #proc_this_action(
+ pool: &action_system::action_pool::ActionPool,
+ mut ctx: action_system::action::ActionContext,
+ #arg_param_name: #arg_type
+ ) -> Result<#return_type, tcp_connection::error::TcpTargetError> {
+ ctx.set_is_remote_action(!#_is_local);
+ 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,
+ args_json,
+ ).await?;
+ serde_json::from_str(&result_json)
+ .map_err(|e| {
+ tcp_connection::error::TcpTargetError::Serialization(e.to_string())
+ })
+ }
+
+ #[allow(dead_code)]
+ #[deprecated = "This function is used by #[action_gen] as a template."]
+ #[doc = "Template function for #[[action_gen]] - do not call directly."]
+ #[doc = "Use the generated struct instead."]
+ #[doc = ""]
+ #[doc = "Register the action to the pool."]
+ #[doc = "```ignore"]
+ #[doc = "register_your_func(&mut pool);"]
+ #[doc = "```"]
+ #[doc = ""]
+ #[doc = "Process the action at the pool."]
+ #[doc = "```ignore"]
+ #[doc = "let result = proc_your_func(&pool, ctx, arg).await?;"]
+ #[doc = "```"]
+ #fn_vis #fn_sig #fn_block
+ }
+}
+
+fn validate_function_signature(fn_sig: &syn::Signature) {
+ if fn_sig.asyncness.is_none() {
+ panic!("Expected async function for Action, but found synchronous function");
+ }
+
+ if fn_sig.inputs.len() != 2 {
+ panic!(
+ "Expected exactly 2 arguments for Action function: ctx: ActionContext and arg: T, but found {} arguments",
+ fn_sig.inputs.len()
+ );
+ }
+
+ let return_type = match &fn_sig.output {
+ syn::ReturnType::Type(_, ty) => ty,
+ _ => panic!(
+ "Expected Action function to return Result<T, TcpTargetError>, but found no return type"
+ ),
+ };
+
+ if let syn::Type::Path(type_path) = return_type.as_ref() {
+ if let Some(segment) = type_path.path.segments.last()
+ && segment.ident != "Result"
+ {
+ panic!(
+ "Expected Action function to return Result<T, TcpTargetError>, but found different return type"
+ );
+ }
+ } else {
+ panic!(
+ "Expected Action function to return Result<T, TcpTargetError>, but found no return type"
+ );
+ }
+}
+
+fn convert_to_pascal_case(s: &str) -> String {
+ s.split('_')
+ .map(|word| {
+ let mut chars = word.chars();
+ match chars.next() {
+ None => String::new(),
+ Some(first) => first.to_uppercase().collect::<String>() + chars.as_str(),
+ }
+ })
+ .collect()
+}
+
+fn extract_parameters_and_types(
+ fn_sig: &syn::Signature,
+) -> (
+ proc_macro2::TokenStream,
+ proc_macro2::TokenStream,
+ proc_macro2::TokenStream,
+ proc_macro2::TokenStream,
+) {
+ let mut inputs = fn_sig.inputs.iter();
+
+ let context_param = match inputs.next() {
+ Some(syn::FnArg::Typed(pat_type)) => {
+ let pat = &pat_type.pat;
+ quote::quote!(#pat)
+ }
+ _ => {
+ panic!("Expected the first argument to be a typed parameter, but found something else")
+ }
+ };
+
+ let arg_param = match inputs.next() {
+ Some(syn::FnArg::Typed(pat_type)) => {
+ let pat = &pat_type.pat;
+ let ty = &pat_type.ty;
+ (quote::quote!(#pat), quote::quote!(#ty))
+ }
+ _ => {
+ panic!("Expected the second argument to be a typed parameter, but found something else")
+ }
+ };
+
+ let (arg_param_name, arg_type) = arg_param;
+
+ let return_type = match &fn_sig.output {
+ syn::ReturnType::Type(_, ty) => {
+ if let syn::Type::Path(type_path) = ty.as_ref() {
+ if let syn::PathArguments::AngleBracketed(args) =
+ &type_path.path.segments.last().unwrap().arguments
+ {
+ if let Some(syn::GenericArgument::Type(ty)) = args.args.first() {
+ quote::quote!(#ty)
+ } else {
+ panic!("Expected to extract the success type of Result, but failed");
+ }
+ } else {
+ panic!("Expected Result type to have generic parameters, but found none");
+ }
+ } else {
+ panic!("Expected return type to be Result, but found different type");
+ }
+ }
+ _ => panic!("Expected function to have return type, but found none"),
+ };
+
+ (context_param, arg_param_name, arg_type, return_type)
+}
diff --git a/systems/action/src/action.rs b/systems/action/src/action.rs
new file mode 100644
index 0000000..62425ff
--- /dev/null
+++ b/systems/action/src/action.rs
@@ -0,0 +1,244 @@
+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, sync::Mutex};
+
+/// # Trait - Action<Args, Return>
+///
+/// A trait used to describe the interaction pattern between client and server
+///
+/// ## Generics
+///
+/// Args: Represents the parameter type required for this action
+///
+/// Return: Represents the return type of this action
+///
+/// The above generics must implement serde's Serialize and DeserializeOwned traits,
+/// and must be sendable between threads
+///
+/// ## Implementation
+///
+/// ```ignore
+/// pub trait Action<Args, Return>
+/// where
+/// Args: Serialize + DeserializeOwned + Send,
+/// Return: Serialize + DeserializeOwned + Send,
+/// {
+/// /// Name, used to inform the server which action to execute
+/// fn action_name() -> &'static str;
+///
+/// /// Whether it's a local Action, used to inform the system if it only runs locally
+/// fn is_remote_action() -> bool;
+///
+/// /// Action processing logic
+/// fn process(
+/// context: ActionContext,
+/// args: Args,
+/// ) -> impl std::future::Future<Output = Result<Return, TcpTargetError>> + Send;
+/// }
+/// ```
+pub trait Action<Args, Return>
+where
+ Args: Serialize + DeserializeOwned + Send,
+ Return: Serialize + DeserializeOwned + Send,
+{
+ fn action_name() -> &'static str;
+
+ fn is_remote_action() -> bool;
+
+ fn process(
+ context: ActionContext,
+ args: Args,
+ ) -> impl std::future::Future<Output = Result<Return, TcpTargetError>> + Send;
+}
+
+/// # Struct - ActionContext
+///
+/// Used to inform the Action about the current execution environment
+///
+/// ## Creation
+///
+/// Create ActionContext using the following methods:
+///
+/// ```ignore
+///
+/// // The instance here is the connection instance passed from external sources for communicating with the server
+/// // For specific usage, please refer to the `/crates/utils/tcp_connection` section
+///
+/// fn init_local_action_ctx(instance: ConnectionInstance) {
+/// // Create context and specify execution on local
+/// let mut ctx = ActionContext::local();
+/// }
+///
+/// fn init_remote_action_ctx(instance: ConnectionInstance) {
+/// // Create context and specify execution on remote
+/// let mut ctx = ActionContext::remote();
+/// }
+#[derive(Default)]
+pub struct ActionContext {
+ /// Whether the action is executed locally or remotely
+ proc_on_local: bool,
+
+ /// Whether the action being executed in the current context is a remote action
+ is_remote_action: 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,
+ instance: Option<Arc<Mutex<ConnectionInstance>>>,
+
+ /// Generic data storage for arbitrary types
+ data: HashMap<TypeId, Arc<dyn Any + Send + Sync>>,
+}
+
+impl ActionContext {
+ /// Generate local context
+ pub fn local() -> Self {
+ ActionContext {
+ proc_on_local: true,
+ ..Default::default()
+ }
+ }
+
+ /// Generate remote context
+ pub fn remote() -> Self {
+ ActionContext {
+ proc_on_local: false,
+ ..Default::default()
+ }
+ }
+
+ /// Build connection instance from TcpStream
+ pub fn build_instance(mut self, stream: TcpStream) -> Self {
+ 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(Arc::new(Mutex::new(instance)));
+ self
+ }
+
+ /// Pop connection instance from context
+ pub fn pop_instance(&mut self) -> Option<Arc<Mutex<ConnectionInstance>>> {
+ self.instance.take()
+ }
+}
+
+impl ActionContext {
+ /// Whether the action is executed locally
+ pub fn is_proc_on_local(&self) -> bool {
+ self.proc_on_local
+ }
+
+ /// Whether the action is executed remotely
+ pub fn is_proc_on_remote(&self) -> bool {
+ !self.proc_on_local
+ }
+
+ /// Whether the action being executed in the current context is a remote action
+ pub fn is_remote_action(&self) -> bool {
+ self.is_remote_action
+ }
+
+ /// Set whether the action being executed in the current context is a remote action
+ pub fn set_is_remote_action(&mut self, is_remote_action: bool) {
+ self.is_remote_action = is_remote_action;
+ }
+
+ /// Get the connection instance in the current context
+ pub fn instance(&self) -> &Option<Arc<Mutex<ConnectionInstance>>> {
+ &self.instance
+ }
+
+ /// Get a mutable reference to the connection instance in the current context
+ pub fn instance_mut(&mut self) -> &mut Option<Arc<Mutex<ConnectionInstance>>> {
+ &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(mut self, action_args: String) -> Self {
+ self.action_args_json = action_args;
+ self
+ }
+
+ /// Insert arbitrary data in the context
+ pub fn with_data<T: Any + Send + Sync>(mut self, value: T) -> Self {
+ self.data.insert(TypeId::of::<T>(), Arc::new(value));
+ self
+ }
+
+ /// Insert arbitrary data as Arc in the context
+ pub fn with_arc_data<T: Any + Send + Sync>(mut self, value: Arc<T>) -> Self {
+ self.data.insert(TypeId::of::<T>(), value);
+ self
+ }
+
+ /// Insert arbitrary data in the context
+ pub fn insert_data<T: Any + Send + Sync>(&mut self, value: T) {
+ self.data.insert(TypeId::of::<T>(), Arc::new(value));
+ }
+
+ /// Insert arbitrary data as Arc in the context
+ pub fn insert_arc_data<T: Any + Send + Sync>(&mut self, value: Arc<T>) {
+ self.data.insert(TypeId::of::<T>(), value);
+ }
+
+ /// Get arbitrary data from the context
+ pub fn get<T: Any + Send + Sync>(&self) -> Option<&T> {
+ self.data
+ .get(&TypeId::of::<T>())
+ .and_then(|arc| arc.downcast_ref::<T>())
+ }
+
+ /// Get arbitrary data as Arc from the context
+ pub fn get_arc<T: Any + Send + Sync>(&self) -> Option<Arc<T>> {
+ self.data
+ .get(&TypeId::of::<T>())
+ .and_then(|arc| Arc::clone(arc).downcast::<T>().ok())
+ }
+
+ /// Remove and return arbitrary data from the context
+ pub fn remove<T: Any + Send + Sync>(&mut self) -> Option<Arc<T>> {
+ self.data
+ .remove(&TypeId::of::<T>())
+ .and_then(|arc| arc.downcast::<T>().ok())
+ }
+
+ /// Check if the context contains data of a specific type
+ pub fn contains<T: Any + Send + Sync>(&self) -> bool {
+ self.data.contains_key(&TypeId::of::<T>())
+ }
+
+ /// Take ownership of the context and extract data of a specific type
+ pub fn take<T: Any + Send + Sync>(mut self) -> (Self, Option<Arc<T>>) {
+ let value = self
+ .data
+ .remove(&TypeId::of::<T>())
+ .and_then(|arc| arc.downcast::<T>().ok());
+ (self, value)
+ }
+}
diff --git a/systems/action/src/action_pool.rs b/systems/action/src/action_pool.rs
new file mode 100644
index 0000000..019fa6d
--- /dev/null
+++ b/systems/action/src/action_pool.rs
@@ -0,0 +1,247 @@
+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 mut ActionContext,
+ args: &'a (dyn std::any::Any + Send + Sync),
+) -> ProcBeginFuture<'a>;
+type ProcEndCallback = fn() -> ProcEndFuture;
+
+type ProcBeginFuture<'a> = Pin<Box<dyn Future<Output = Result<(), TcpTargetError>> + Send + 'a>>;
+type ProcEndFuture = Pin<Box<dyn Future<Output = Result<(), TcpTargetError>> + Send>>;
+
+/// # Struct - ActionPool
+///
+/// This struct is used to register and record all accessible and executable actions
+///
+/// It also registers `on_proc_begin` and `on_proc_end` callback functions
+/// used for action initialization
+///
+/// ## Creating and registering actions
+/// ```ignore
+/// fn init_action_pool() {
+/// let mut pool = Action::new();
+///
+/// // Register action
+/// pool.register<YourAction, ActionArgument, ActionReturn>();
+///
+/// // If the action is implemented with `#[action_gen]`, you can also do
+/// register_your_action(&mut pool);
+/// }
+/// ```
+pub struct ActionPool {
+ /// HashMap storing action name to action implementation mapping
+ actions: std::collections::HashMap<&'static str, Box<dyn ActionErased>>,
+
+ /// Callback to execute when process begins
+ on_proc_begin: Option<ProcBeginCallback>,
+
+ /// Callback to execute when process ends
+ on_proc_end: Option<ProcEndCallback>,
+}
+
+impl Default for ActionPool {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl ActionPool {
+ /// Creates a new empty 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::<MyAction, MyArgs, MyReturn>();
+ /// ```
+ pub fn register<A, Args, Return>(&mut self)
+ where
+ A: Action<Args, Return> + 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(
+ action_name,
+ Box::new(ActionWrapper::<A, Args, Return>(std::marker::PhantomData)),
+ );
+ }
+
+ /// Processes an action by name with given context and arguments
+ ///
+ /// Usage:
+ /// ```ignore
+ /// let result = action_pool.process::<MyArgs, MyReturn>("my_action", context, args).await?;
+ /// ```
+ /// 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(args_json.clone());
+
+ self.exec_on_proc_begin(&mut context, &args_json).await?;
+ let result = action.process_json_erased(context, args_json).await?;
+ 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 + Sync + 'static,
+ Return: serde::Serialize + Send + 'static,
+ {
+ if let Some(action) = self.actions.get(action_name) {
+ self.exec_on_proc_begin(&mut context, &args).await?;
+ let result = action.process_erased(context, Box::new(args)).await?;
+ let result = *result
+ .downcast::<Return>()
+ .map_err(|_| TcpTargetError::Unsupported("InvalidArguments".to_string()))?;
+ 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: &mut ActionContext,
+ args: &(dyn std::any::Any + Send + Sync),
+ ) -> Result<(), TcpTargetError> {
+ if let Some(callback) = &self.on_proc_begin {
+ callback(context, args).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
+type ProcessErasedFuture = std::pin::Pin<
+ Box<
+ dyn std::future::Future<Output = Result<Box<dyn std::any::Any + Send>, TcpTargetError>>
+ + Send,
+ >,
+>;
+type ProcessJsonErasedFuture =
+ std::pin::Pin<Box<dyn std::future::Future<Output = Result<String, TcpTargetError>> + 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<dyn std::any::Any + Send>,
+ ) -> ProcessErasedFuture;
+
+ /// Processes the action with JSON-serialized arguments and returns JSON-serialized result
+ fn process_json_erased(
+ &self,
+ context: ActionContext,
+ args_json: String,
+ ) -> ProcessJsonErasedFuture;
+}
+
+/// Wrapper struct that implements ActionErased for concrete Action types
+struct ActionWrapper<A, Args, Return>(std::marker::PhantomData<(A, Args, Return)>);
+
+impl<A, Args, Return> ActionErased for ActionWrapper<A, Args, Return>
+where
+ A: Action<Args, Return> + Send + Sync,
+ Args: Serialize + DeserializeOwned + Send + Sync + 'static,
+ Return: Serialize + DeserializeOwned + Send + Sync + 'static,
+{
+ fn process_erased(
+ &self,
+ context: ActionContext,
+ args: Box<dyn std::any::Any + Send>,
+ ) -> std::pin::Pin<
+ Box<
+ dyn std::future::Future<Output = Result<Box<dyn std::any::Any + Send>, TcpTargetError>>
+ + Send,
+ >,
+ > {
+ Box::pin(async move {
+ let args = *args
+ .downcast::<Args>()
+ .map_err(|_| TcpTargetError::Unsupported("InvalidArguments".to_string()))?;
+ let result = A::process(context, args).await?;
+ 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)
+ })
+ }
+}
diff --git a/systems/action/src/lib.rs b/systems/action/src/lib.rs
new file mode 100644
index 0000000..12ae999
--- /dev/null
+++ b/systems/action/src/lib.rs
@@ -0,0 +1,6 @@
+pub mod macros {
+ pub use action_system_macros::*;
+}
+
+pub mod action;
+pub mod action_pool;
generated by cgit (git 2.34.1) at 2026-03-23 16:22:04 +0000