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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

mod vec_stream;

use std::{future::ready, pin::Pin, time::Duration};

use futures::{
    stream::{self, BoxStream},
    FutureExt, Stream, StreamExt,
};
use tokio::time;
use tokio_stream::wrappers::IntervalStream;
use vec_stream::VecStreamExt;

use super::{OutputBuffer, TaskTransform};
use crate::event::Event;

/// A structure representing user-defined timer.
#[derive(Clone, Copy, Debug)]
pub struct Timer {
    pub id: u32,
    pub interval: Duration,
}

/// A trait representing a runtime running user-defined code.
pub trait RuntimeTransform {
    /// Call user-defined "init" hook.
    fn hook_init<F>(&mut self, _emit_fn: F)
    where
        F: FnMut(Event),
    {
    }

    /// Call user-defined "process" hook.
    fn hook_process<F>(&mut self, event: Event, emit_fn: F)
    where
        F: FnMut(Event);

    /// Call user-defined "shutdown" hook.
    fn hook_shutdown<F>(&mut self, _emit_fn: F)
    where
        F: FnMut(Event),
    {
    }

    /// Call user-defined timer handler.
    fn timer_handler<F>(&mut self, _timer: Timer, _emit_fn: F)
    where
        F: FnMut(Event),
    {
    }

    /// Return (static) list of user-defined timers.
    fn timers(&self) -> Vec<Timer> {
        Vec::new()
    }

    fn transform(&mut self, output: &mut OutputBuffer, event: Event) {
        let mut maybe = None;
        self.hook_process(event, |event| maybe = Some(event));
        output.extend(maybe.into_iter());
    }
}

#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
enum Message {
    Init,
    Process(Event),
    Shutdown,
    Timer(Timer),
}

impl<T> TaskTransform<Event> for T
where
    T: RuntimeTransform + Send + 'static,
{
    fn transform(
        mut self: Box<Self>,
        input_rx: Pin<Box<dyn Stream<Item = Event> + Send>>,
    ) -> Pin<Box<dyn Stream<Item = Event> + Send>>
    where
        Self: 'static,
    {
        let timers = self.timers();
        let mut is_shutdown: bool = false; // TODO: consider using an enum describing the state instead of a
                                           // a single boolean variable.
                                           // It is used to prevent timers to emit messages after the source
                                           // stream stopped.

        Box::pin(
            input_rx
                .map(Message::Process)
                .fuse()
                .into_future()
                .map(move |(first, rest)| {
                    // The first message is always `Message::Init`.
                    let init_msg = stream::once(ready(Message::Init));
                    // After it comes the first event, if any.
                    let first_event = first.map_or_else(
                        || stream::empty().boxed(),
                        |msg| stream::once(ready(msg)).boxed(),
                    );
                    // Then all other events followed by `Message::Shutdown` message
                    let rest_events_and_shutdown_msg =
                        rest.chain(stream::once(ready(Message::Shutdown)));
                    // A stream of `Message::Timer(..)` events generated by timers.
                    let timer_msgs = make_timer_msgs_stream(timers);

                    init_msg
                        .chain(first_event)
                        .chain(
                            // We need to finish when `rest_events_and_shutdown_msg` finishes so
                            // not to hang on timers, but not finish when `timer_msgs` finishes
                            // as there may not be any timer.
                            rest_events_and_shutdown_msg
                                .select_weak(timer_msgs.chain(stream::pending())),
                        )
                        .boxed()
                })
                .into_stream()
                .flatten()
                .map(move |msg| {
                    let mut acc = Vec::new(); // TODO: create a stream adaptor to avoid buffering all events
                    if !is_shutdown {
                        match msg {
                            Message::Init => self.hook_init(|event| acc.push(event)),
                            Message::Process(event) => {
                                self.hook_process(event, |event| acc.push(event));
                            }
                            Message::Shutdown => {
                                self.hook_shutdown(|event| acc.push(event));
                                is_shutdown = true;
                            }
                            Message::Timer(timer) => {
                                self.timer_handler(timer, |event| acc.push(event));
                            }
                        }
                    }
                    stream::iter(acc).boxed()
                })
                .flatten()
                .boxed(),
        )
    }
}

fn make_timer_msgs_stream(timers: Vec<Timer>) -> BoxStream<'static, Message> {
    let streams = timers.into_iter().map(|timer| {
        IntervalStream::new(time::interval(timer.interval)).map(move |_| Message::Timer(timer))
    });
    stream::select_all(streams).boxed()
}