1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
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<Box<dyn Future<Output = Result<(), TcpTargetError>> + Send + 'a>>;
type ProcEndCallback = fn() -> Pin<Box<dyn Future<Output = Result<(), TcpTargetError>> + 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<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 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?;
/// ```
pub async fn process<'a, Args, Return>(
&'a self,
action_name: &'a str,
context: ActionContext,
args_json: String,
) -> 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 result = action.process_erased(context, Box::new(args)).await?;
let result = *result
.downcast::<Return>()
.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
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>,
) -> std::pin::Pin<
Box<
dyn std::future::Future<Output = Result<Box<dyn std::any::Any + Send>, TcpTargetError>>
+ Send,
>,
>;
}
/// 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>)
})
}
}
|
