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

extern crate futures;

use std::error::Error;
use std::fmt;

use futures::{Async, Future, Poll, Stream};
use futures::future;
use futures::future::Executor;
use futures::sync::{mpsc, oneshot};

/// Construct a new actor, requires an `Executor` and an initial state.  Returns a reference that can be cheaply
/// cloned and passed between threads.  A specific implementation is expected to wrap this return value and implement
/// the required custom logic.
pub fn actor<EX, A, S, R, E>(executor: &EX, initial_state: S) -> ActorSender<A, R, E>
where
    A: Action<S, R, E> + Send + 'static,
    S: Send + 'static,
    R: Send + 'static,
    E: Send + 'static,
    EX: Executor<Box<Future<Item = (), Error = ()> + Send>> + 'static,
{
    let (tx, rx) = mpsc::unbounded();
    let actor = Box::new(Actor {
        receiver: rx,
        state: initial_state,
    });
    executor
        .execute(actor)
        .expect("Cannot schedule actor on executor");
    ActorSender(tx)
}

type ActorSenderInner<A, R, E> = mpsc::UnboundedSender<(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,
{
    /// Invokes a specific action on the actor.  Returns a future that completes when the actor has
    /// performed the action
    pub fn invoke(&self, action: A) -> ActFuture<R, E> {
        let (tx, rx) = oneshot::channel();
        if let Err(e) = self.0.unbounded_send((action, tx)) {
            return Box::new(future::err(ActorError::from(e).into()));
        }
        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()),
            })
        });
        Box::new(recv_f)
    }
}

struct Actor<A, S, R, E> {
    receiver: mpsc::UnboundedReceiver<(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)))) => {
                    // Not checking the result, as nothing may be waiting
                    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 {
    /// Cannot send message to the actor
    InvokeError,
    /// Response was cancelled before being received.
    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
    }
}