flmodules 0.10.0

Modules used in fledger
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

use std::time::Duration;
use tokio_stream::StreamExt;

use flarch::broker::{Broker, Message};
use flarch::tasks::{now, spawn_local, Interval};

#[derive(Debug, Clone, PartialEq)]
pub enum TimerMessage {
    Second,
    Minute,
}

pub type BrokerTimer = Broker<(), TimerMessage>;

/// The Timer structure sends out periodic signals to the system so that
/// services can subscribe to them.
#[derive(Clone, Debug)]
pub struct Timer {
    pub broker: BrokerTimer,
}

impl Timer {
    pub async fn start() -> anyhow::Result<Timer> {
        let broker = Broker::new();
        let mut broker_cl = broker.clone();
        spawn_local(async move {
            let mut last_tick = now();
            let mut seconds = 0;
            let mut interval = Interval::new_interval(Duration::from_secs(1));
            loop {
                interval.next().await;
                if now() - last_tick > 2000 {
                    log::warn!(
                        "Timer took {}ms instead of 1000ms, this means the machine is too busy!",
                        now() - last_tick
                    );
                }
                last_tick = now();
                seconds += 1;
                broker_cl
                    .emit_msg_out(TimerMessage::Second)
                    .err()
                    .map(|e| log::error!("While emitting timer: {e:?}"));
                if seconds == 60 {
                    broker_cl
                        .emit_msg_out(TimerMessage::Minute)
                        .err()
                        .map(|e| log::error!("While emitting timer: {e:?}"));
                    seconds = 0;
                }
            }
        });
        Ok(Timer { broker })
    }

    pub fn simul() -> Timer {
        Timer {
            broker: Broker::new(),
        }
    }

    pub async fn second<I: Message + 'static, O: Message + 'static>(
        mut timer: BrokerTimer,
        broker: Broker<I, O>,
        msg: I,
    ) -> anyhow::Result<()> {
        timer
            .add_translator_o_ti(
                broker,
                Box::new(move |m| match m {
                    TimerMessage::Second => Some(msg.clone()),
                    TimerMessage::Minute => None,
                }),
            )
            .await?;
        Ok(())
    }

    pub async fn minute<I: Message + 'static, O: Message + 'static>(
        mut timer: BrokerTimer,
        broker: Broker<I, O>,
        msg: I,
    ) -> anyhow::Result<()> {
        timer
            .add_translator_o_ti(
                broker,
                Box::new(move |m| match m {
                    TimerMessage::Second => None,
                    TimerMessage::Minute => Some(msg.clone()),
                }),
            )
            .await?;
        Ok(())
    }

    pub async fn tick<I: Message + 'static, O: Message + 'static>(
        &mut self,
        broker: Broker<I, O>,
        msg: I,
        tick: TimerMessage,
    ) -> anyhow::Result<()> {
        self.broker
            .add_translator_o_ti(
                broker,
                Box::new(move |tm| (tm == tick).then_some(msg.clone())),
            )
            .await?;

        Ok(())
    }

    pub async fn tick_second<I: Message + 'static, O: Message + 'static>(
        &mut self,
        broker: Broker<I, O>,
        msg: I,
    ) -> anyhow::Result<()> {
        self.tick(broker, msg, TimerMessage::Second).await
    }

    pub async fn tick_minute<I: Message + 'static, O: Message + 'static>(
        &mut self,
        broker: Broker<I, O>,
        msg: I,
    ) -> anyhow::Result<()> {
        self.tick(broker, msg, TimerMessage::Minute).await
    }
}