actix 0.5.3

Actor framework for Rust
Documentation 
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

use std::marker::PhantomData;
use std::time::Duration;
use futures::{Async, Future, Poll, Stream};
use tokio_core::reactor::Timeout;

use fut::ActorFuture;
use arbiter::Arbiter;
use actor::{Actor, ActorContext, AsyncContext};
use handler::{Handler, MessageResponse, Message};


pub(crate) struct ActorWaitItem<A: Actor>(Box<ActorFuture<Item=(), Error=(), Actor=A>>);

impl<A> ActorWaitItem<A> where A: Actor, A::Context: ActorContext + AsyncContext<A> {

    #[inline]
    pub fn new<F>(fut: F) -> Self where F: ActorFuture<Item=(), Error=(), Actor=A> + 'static {
        ActorWaitItem(Box::new(fut))
    }

    pub fn poll(&mut self, act: &mut A, ctx: &mut A::Context) -> Async<()> {
        match self.0.poll(act, ctx) {
            Ok(Async::NotReady) => {
                if ctx.state().alive() {
                    Async::NotReady
                } else {
                    Async::Ready(())
                }
            },
            Ok(Async::Ready(_)) | Err(_) => Async::Ready(()),
        }
    }
}

pub(crate)
struct ActorDelayedMessageItem<A, M> where A: Actor, M: Message {
    msg: Option<M>,
    timeout: Timeout,
    act: PhantomData<A>,
    m: PhantomData<M>,
}

impl<A, M> ActorDelayedMessageItem<A, M> where A: Actor, M: Message {
    pub fn new(msg: M, timeout: Duration) -> Self {
        ActorDelayedMessageItem {
            msg: Some(msg),
            timeout: Timeout::new(timeout, Arbiter::handle()).unwrap(),
            act: PhantomData,
            m: PhantomData,
        }
    }
}

impl<A, M> ActorFuture for ActorDelayedMessageItem<A, M>
    where A: Actor + Handler<M>, A::Context: AsyncContext<A>, M: Message + 'static,
{
    type Item = ();
    type Error = ();
    type Actor = A;

    fn poll(&mut self, act: &mut A, ctx: &mut A::Context) -> Poll<Self::Item, Self::Error> {
        match self.timeout.poll() {
            Ok(Async::NotReady) => Ok(Async::NotReady),
            Ok(Async::Ready(_)) => {
                let fut = A::handle(act, self.msg.take().unwrap(), ctx);
                fut.handle::<()>(ctx, None);
                Ok(Async::Ready(()))
            },
            Err(_) => unreachable!(),
        }
    }
}

pub(crate)
struct ActorMessageItem<A, M> where A: Actor, M: Message {
    msg: Option<M>,
    act: PhantomData<A>,
}

impl<A, M> ActorMessageItem<A, M> where A: Actor, M: Message {
    pub fn new(msg: M) -> Self {
        ActorMessageItem{msg: Some(msg), act: PhantomData}
    }
}

impl<A, M: 'static> ActorFuture for ActorMessageItem<A, M>
    where A: Actor + Handler<M>, A::Context: AsyncContext<A>, M: Message
{
    type Item = ();
    type Error = ();
    type Actor = A;

    fn poll(&mut self, act: &mut A, ctx: &mut A::Context) -> Poll<Self::Item, Self::Error> {
        let fut = Handler::handle(act, self.msg.take().unwrap(), ctx);
        fut.handle::<()>(ctx, None);
        Ok(Async::Ready(()))
    }
}

pub(crate)
struct ActorMessageStreamItem<A, M, S> where A: Actor, M: Message {
    stream: S,
    act: PhantomData<A>,
    msg: PhantomData<M>,
}

impl<A, M, S> ActorMessageStreamItem<A, M, S> where A: Actor, M: Message {
    pub fn new(st: S) -> Self {
        ActorMessageStreamItem { stream: st, act: PhantomData, msg: PhantomData }
    }
}

impl<A, M: 'static, S> ActorFuture for ActorMessageStreamItem<A, M, S>
    where S: Stream<Item=M, Error=()>,
          A: Actor + Handler<M>, A::Context: AsyncContext<A>,
          M: Message
{
    type Item = ();
    type Error = ();
    type Actor = A;

    fn poll(&mut self, act: &mut A, ctx: &mut A::Context) -> Poll<Self::Item, Self::Error> {
        loop {
            match self.stream.poll() {
                Ok(Async::Ready(Some(msg))) => {
                    let fut = Handler::handle(act, msg, ctx);
                    fut.handle::<()>(ctx, None);
                    if ctx.waiting() {
                        return Ok(Async::NotReady)
                    }
                }
                Ok(Async::Ready(None)) => return Ok(Async::Ready(())),
                Ok(Async::NotReady) => return Ok(Async::NotReady),
                Err(_) => (),
            }
        }
    }
}