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

use std::time::Duration;
use std::marker::PhantomData;

use futures::{Async, Future, Poll};
use tokio_core::reactor::Timeout;

use arbiter::Arbiter;
use handler::{Handler, Message};

use super::{ToEnvelope, SendError, MailboxError};
use super::{MessageDestination, MessageDestinationTransport};


/// `Request` is a `Future` which represents asynchronous message sending process.
#[must_use = "You have to wait on request otherwise Message wont be delivered"]
pub struct Request<T, A, M>
    where T: MessageDestination<A, M>,
          T::Transport: MessageDestinationTransport<T, A, M>,
          A: Handler<M>, A::Context: ToEnvelope<T, A, M>,
          M: Message + 'static,
{
    rx: Option<T::ResultReceiver>,
    info: Option<(T::Transport, M)>,
    timeout: Option<Timeout>,
    act: PhantomData<A>,
}

impl<T, A, M> Request<T, A, M>
    where T: MessageDestination<A, M>,
          T::Transport: MessageDestinationTransport<T, A, M>,
          A: Handler<M>, A::Context: ToEnvelope<T, A, M>,
          M: Message + 'static,
{
    pub(crate) fn new(rx: Option<T::ResultReceiver>,
                      info: Option<(T::Transport, M)>) -> Request<T, A, M> {
        Request{rx, info, timeout: None, act: PhantomData}
    }

    /// Set message delivery timeout
    pub fn timeout(mut self, dur: Duration) -> Self {
        self.timeout = Some(Timeout::new(dur, Arbiter::handle()).unwrap());
        self
    }

    fn poll_timeout(&mut self) -> Poll<M::Result, MailboxError> {
        if let Some(ref mut timeout) = self.timeout {
            match timeout.poll() {
                Ok(Async::Ready(())) => Err(MailboxError::Timeout),
                Ok(Async::NotReady) => Ok(Async::NotReady),
                Err(_) => unreachable!()
            }
        } else {
            Ok(Async::NotReady)
        }
    }
}

impl<T, A, M> Future for Request<T, A, M>
    where T: MessageDestination<A, M>,
          T::Transport: MessageDestinationTransport<T, A, M>,
          A: Handler<M>, A::Context: ToEnvelope<T, A, M>, M: Message + 'static,
{
    type Item = M::Result;
    type Error = MailboxError;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        if let Some((sender, msg)) = self.info.take() {
            match sender.send(msg) {
                Ok(rx) => self.rx = Some(rx),
                Err(SendError::Full(msg)) => {
                    self.info = Some((sender, msg));
                    return Ok(Async::NotReady)
                }
                Err(SendError::Closed(_)) => return Err(MailboxError::Closed),
            }
        }

        if let Some(mut rx) = self.rx.take() {
            match rx.poll() {
                Ok(Async::Ready(item)) => Ok(Async::Ready(item)),
                Ok(Async::NotReady) => {
                    self.rx = Some(rx);
                    self.poll_timeout()
                }
                Err(_) => Err(MailboxError::Closed),
            }
        } else {
            Err(MailboxError::Closed)
        }
    }
}