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

extern crate actix;
extern crate futures;
extern crate tokio_core;

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use futures::{Future, Stream};
use futures::stream::futures_ordered;
use tokio_core::reactor::Timeout;
use actix::prelude::*;
use actix::msgs::SystemExit;

struct MyActor {
    timeout: Arc<AtomicBool>,
}

#[derive(PartialEq, Copy, Clone)]
enum Error {
    Timeout,
    Generic,
}

impl Actor for MyActor {
    type Context = actix::Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        Timeout::new(Duration::new(0, 5_000_000), Arbiter::handle()).unwrap()
            .then(|_| {
                Arbiter::system().do_send(SystemExit(0));
                Ok::<_, Error>(())
            })
            .into_actor(self)
            .timeout(Duration::new(0, 100), Error::Timeout)
            .map_err(|e, act, _| if e == Error::Timeout {
                act.timeout.store(true, Ordering::Relaxed);
                Arbiter::system().do_send(SystemExit(0));
                ()
            })
            .wait(ctx)
    }
}

#[test]
fn test_fut_timeout() {
    let sys = System::new("test");
    let timeout = Arc::new(AtomicBool::new(false));

    let _addr: Addr<Unsync, _> = MyActor {timeout: Arc::clone(&timeout)}.start();

    sys.run();
    assert!(timeout.load(Ordering::Relaxed), "Not timeout");
}


struct MyStreamActor {
    timeout: Arc<AtomicBool>,
}

impl Actor for MyStreamActor {
    type Context = actix::Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        let s = futures_ordered(
            vec![Timeout::new(Duration::new(0, 5_000_000), Arbiter::handle()),
                 Timeout::new(Duration::new(0, 5_000_000), Arbiter::handle())]);

        s.and_then(|f| f)
            .map_err(|_| Error::Generic)
            .into_actor(self)
            .timeout(Duration::new(0, 100), Error::Timeout)
            .map_err(|e, act, _| if e == Error::Timeout {
                act.timeout.store(true, Ordering::Relaxed);
                Arbiter::system().do_send(SystemExit(0));
                ()
            })
            .finish()
            .wait(ctx)
    }
}

#[test]
fn test_stream_timeout() {
    let sys = System::new("test");
    let timeout = Arc::new(AtomicBool::new(false));

    let _addr: Addr<Unsync, _> = MyStreamActor {timeout: Arc::clone(&timeout)}.start();

    sys.run();
    assert!(timeout.load(Ordering::Relaxed), "Not timeout");
}