actix 0.5.7

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

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

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

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, 1000), 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");
}