coerce 0.8.6

Async actor runtime and distributed systems framework
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

use crate::actor::message::{Handler, Message};
use crate::actor::{Actor, LocalActorRef};
use std::ops::Add;
use tracing::trace;

use std::time::Duration;
use tokio::sync::oneshot;
use tokio::time;
use tokio::time::Instant;
use uuid::Uuid;

pub trait TimerTick: Message {}

enum TimerMode {
    Notify,
    Send,
}

pub struct Timer {
    stop: oneshot::Sender<bool>,
}

impl Timer {
    pub fn start_immediately<A: Actor, T: TimerTick>(
        actor: LocalActorRef<A>,
        tick: Duration,
        msg: T,
    ) -> Timer
    where
        A: 'static + Handler<T> + Sync + Send,
        T: 'static + Clone + Sync + Send,
        T::Result: 'static + Sync + Send,
    {
        let (stop, stop_rx) = oneshot::channel();
        tokio::spawn(timer_loop(tick, msg, actor, stop_rx, true, TimerMode::Send));

        Timer { stop }
    }

    pub fn start<A: Actor, T: TimerTick>(actor: LocalActorRef<A>, tick: Duration, msg: T) -> Timer
    where
        A: 'static + Handler<T> + Sync + Send,
        T: 'static + Clone + Sync + Send,
        T::Result: 'static + Sync + Send,
    {
        let (stop, stop_rx) = oneshot::channel();
        tokio::spawn(timer_loop(
            tick,
            msg,
            actor,
            stop_rx,
            false,
            TimerMode::Send,
        ));

        Timer { stop }
    }

    pub fn stop(self) -> bool {
        if self.stop.send(true).is_ok() {
            true
        } else {
            false
        }
    }
}

async fn timer_loop<A: Actor, T: TimerTick>(
    tick: Duration,
    msg: T,
    actor: LocalActorRef<A>,
    mut stop_rx: oneshot::Receiver<bool>,
    tick_immediately: bool,
    mode: TimerMode,
) where
    A: Handler<T>,
    T: 'static + Clone + Sync + Send,
{
    let start = if tick_immediately {
        Instant::now()
    } else {
        Instant::now().add(tick)
    };

    let mut interval = time::interval_at(start, tick);
    let timer_id = Uuid::new_v4();

    interval.tick().await;

    trace!("{} - timer starting", &timer_id);

    loop {
        if stop_rx.try_recv().is_ok() {
            break;
        }

        trace!("{} - timer tick", &timer_id);

        let now = Instant::now();

        match mode {
            TimerMode::Notify => {
                if actor.notify(msg.clone()).is_err() {
                    break;
                }
            }
            TimerMode::Send => {
                if actor.send(msg.clone()).await.is_err() {
                    break;
                }

                interval.reset();
            }
        }

        trace!(
            "{} - tick res received in {}ms",
            &timer_id,
            now.elapsed().as_millis()
        );
        interval.tick().await;
    }

    trace!("{} - timer finished", timer_id);
}