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
use std::{future::Future, marker::PhantomData};
use crate::{Actor, Ask, AskResult, AsyncAsk, Ctx, Handler, Message};
pub struct AnonymousActor<In: Message, Out: Message, F: Fn(In) -> Out> {
f: Option<F>,
_in: PhantomData<In>,
_out: PhantomData<Out>,
}
impl<In, Out, F> From<F> for AnonymousActor<In, Out, F>
where
In: Message,
Out: Message,
F: Fn(In) -> Out + Send + Sync + 'static,
{
fn from(f: F) -> Self {
Self {
f: Some(f),
_in: PhantomData,
_out: PhantomData,
}
}
}
impl<In, Out, F> Actor for AnonymousActor<In, Out, F>
where
In: Message,
Out: Message,
F: Fn(In) -> Out + Send + Sync + 'static,
{
fn name() -> &'static str {
"AnoymousFnActor"
}
fn mailbox_size() -> usize {
1
}
}
impl<In, F> Handler<In> for AnonymousActor<In, (), F>
where
In: Message,
F: Fn(In) + Send + Sync + 'static,
{
fn handle(&mut self, message: In, _: &mut Ctx<Self>) {
let f = self.f.take().unwrap();
(f)(message);
}
}
impl<In, Out, F> Ask<In> for AnonymousActor<In, Out, F>
where
In: Message,
Out: Message,
F: Fn(In) -> Out + Send + Sync + 'static,
{
type Result = Out;
fn handle(&mut self, message: In, _: &mut Ctx<Self>) -> AskResult<Out> {
let f = self.f.take().unwrap();
AskResult::Reply((f)(message))
}
}
impl<In, Out, Fut, F> AsyncAsk<In> for AnonymousActor<In, Fut, F>
where
In: Message,
Out: Message,
F: Fn(In) -> Fut + Send + Sync + 'static,
for<'a> Fut: Future<Output = Out> + Send + Sync + 'a,
{
type Output = Out;
type Future<'a> = Fut;
fn handle<'a>(&'a mut self, message: In, _: &mut Ctx<Self>) -> Self::Future<'a> {
let f = self.f.take().unwrap();
(f)(message)
}
}
#[cfg(test)]
mod test {
// use crate::{util::Workflow, utils::workflow::WorkflowBase, Actor, Ask, Message};
// struct Response {
// rx: tokio::sync::oneshot::Receiver<Increment>,
// }
// struct Increment(usize);
// async fn increment(msg: Increment) -> Increment {
// Increment(msg.0 + 1)
// }
// use crate::{Actor, Handler, Message};
// trait All: Send + Sync + 'static {}
// struct GenericRunner<A: All, B: All, C: All> {
// a: A,
// b: B,
// c: C,
// }
// impl<A: All, B: All, C: All> Actor for GenericRunner<A, B, C> {}
// impl<A: All + Message, B: All, C: All> Handler<A> for GenericRunner<A, B, C> {
// fn handle(&mut self, message: A, context: &mut crate::Ctx<Self>) {
// println!("Get a message for A")
// }
// }
// impl Actor for Runner {}
// impl Ask<Increment> for Runner {
// type Result = Response;
// fn handle(&mut self, message: Increment, context: &mut crate::Ctx<Self>) -> Self::Result {
// let (tx, rx) = tokio::sync::oneshot::channel();
// let future = increment
// .then(increment)
// .then(increment)
// .then(increment)
// .then(increment)
// .run(message);
// let handle = context.anonymous(future);
// }
// }
// #[tokio::test]
// async fn test() {
// let runner = Runner;
// let address = runner.start();
// address.try_send(Increment(0));
// let output = assert_eq!(output.0, 5);
// }
}