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
extern crate futures;
extern crate futures_cpupool;
use std::error::Error;
use std::fmt;
use futures::{Async, Future, Poll, Sink, Stream};
use futures::future;
use futures::future::Executor;
use futures::sync::{mpsc, oneshot};
use futures_cpupool::CpuPool;
const DEFAULT_SIZE: usize = 8;
pub fn actor<A, S, R, E>(cpu_pool: &CpuPool, initial_state: S) -> ActorSender<A, R, E>
where
A: Action<S, R, E> + Send + 'static,
S: Send + 'static,
R: Send + 'static,
E: Send + 'static,
{
let (tx, rx) = mpsc::channel(DEFAULT_SIZE);
let actor = Actor {
receiver: rx,
state: initial_state,
};
cpu_pool
.execute(actor)
.expect("Cannot schedule actor on executor");
ActorSender(tx)
}
type ActorSenderInner<A, R, E> = mpsc::Sender<(A, oneshot::Sender<Result<R, E>>)>;
#[derive(Debug)]
pub struct ActorSender<A, R, E>(ActorSenderInner<A, R, E>);
impl<A, R, E> Clone for ActorSender<A, R, E> {
fn clone(&self) -> Self {
ActorSender(self.0.clone())
}
}
impl<A, R, E> ActorSender<A, R, E>
where
A: Send + 'static,
R: Send + 'static,
E: Send + From<ActorError> + 'static,
{
pub fn invoke(&self, action: A) -> ActFuture<R, E> {
let (tx, rx) = oneshot::channel();
let send_f = self.0.clone().send((action, tx));
let recv_f = rx.then(|r| {
future::result(match r {
Ok(Ok(r)) => Ok(r),
Ok(Err(e)) => Err(e),
Err(e) => Err(ActorError::from(e).into()),
})
});
let act_f = send_f.map_err(|e| ActorError::from(e).into()).join(recv_f);
Box::new(act_f.map(|(_, result)| result))
}
}
struct Actor<A, S, R, E> {
receiver: mpsc::Receiver<(A, oneshot::Sender<Result<R, E>>)>,
state: S,
}
impl<A, S, R, E> Future for Actor<A, S, R, E>
where
A: Action<S, R, E>,
{
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
loop {
match self.receiver.poll() {
Ok(Async::Ready(Some((a, tx)))) => {
let _ = tx.send(a.act(&mut self.state));
}
Ok(Async::Ready(None)) => return Ok(Async::Ready(())),
Ok(Async::NotReady) => return Ok(Async::NotReady),
Err(()) => return Err(()),
}
}
}
}
pub trait Action<S, R, E> {
fn act(self, s: &mut S) -> Result<R, E>;
}
pub type ActFuture<R, E> = Box<Future<Item = R, Error = E> + Send>;
#[derive(Debug)]
pub enum ActorError {
InvokeError,
WaitError,
}
impl Error for ActorError {
fn description(&self) -> &str {
match self {
&ActorError::InvokeError => "Cannot send message to actor",
&ActorError::WaitError => "Cannot wait for an answer",
}
}
}
impl fmt::Display for ActorError {
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
write!(f, "{}", self.description())
}
}
impl<T> From<mpsc::SendError<T>> for ActorError {
fn from(_from: mpsc::SendError<T>) -> Self {
ActorError::InvokeError
}
}
impl From<oneshot::Canceled> for ActorError {
fn from(_from: oneshot::Canceled) -> Self {
ActorError::WaitError
}
}