p2panda_stream/
hooks.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3use std::cell::RefCell;
4use std::collections::VecDeque;
5use std::convert::Infallible;
6use std::pin::Pin;
7
8use tokio::sync::Notify;
9
10use crate::Processor;
11
12/// Hook to trigger custom actions by observing processor events passing through the pipeline.
13pub trait ProcessorHook<T>: Send {
14    fn on_input<'a>(&'a self, input: &'a T) -> impl Future<Output = ()> + Send + 'a;
15}
16
17type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
18
19trait DynProcessorHook<T>: Send {
20    fn on_input<'a>(&'a self, input: &'a T) -> BoxFuture<'a, ()>;
21}
22
23impl<I, T: ProcessorHook<I>> DynProcessorHook<I> for T {
24    fn on_input<'a>(&'a self, input: &'a I) -> BoxFuture<'a, ()> {
25        Box::pin(ProcessorHook::on_input(self, input))
26    }
27}
28
29/// List of custom hooks which can be registered on a single processor layer.
30#[derive(Default)]
31pub struct ProcessorHooksList<T> {
32    inner: Vec<Box<dyn DynProcessorHook<T> + Sync>>,
33}
34
35impl<T> ProcessorHooksList<T> {
36    pub fn new() -> Self {
37        Self { inner: Vec::new() }
38    }
39
40    pub fn push(&mut self, hook: impl ProcessorHook<T> + Sync + 'static) {
41        self.inner.push(Box::new(hook));
42    }
43}
44
45impl<T> ProcessorHook<T> for ProcessorHooksList<T>
46where
47    T: Sync,
48{
49    async fn on_input(&self, input: &T) {
50        for hook in &self.inner {
51            hook.on_input(input).await;
52        }
53    }
54}
55
56#[derive(Default)]
57pub struct Hooks<T> {
58    list: ProcessorHooksList<T>,
59    notify: Notify,
60    queue: RefCell<VecDeque<T>>,
61}
62
63impl<T> Hooks<T> {
64    pub fn new() -> Self {
65        Self::from_list(ProcessorHooksList::new())
66    }
67
68    pub fn from_list(list: ProcessorHooksList<T>) -> Self {
69        Self {
70            list,
71            notify: Notify::new(),
72            queue: RefCell::new(VecDeque::new()),
73        }
74    }
75
76    pub fn push(&mut self, hook: impl ProcessorHook<T> + Sync + 'static) {
77        self.list.push(hook);
78    }
79}
80
81impl<T> Processor<T> for Hooks<T>
82where
83    T: Sync,
84{
85    type Output = T;
86
87    type Error = Infallible;
88
89    async fn process(&self, input: T) -> Result<(), Self::Error> {
90        ProcessorHook::on_input(&self.list, &input).await;
91
92        self.queue.borrow_mut().push_back(input);
93        self.notify.notify_one();
94
95        Ok(())
96    }
97
98    async fn next(&self) -> Result<Self::Output, Self::Error> {
99        loop {
100            if let Some(output) = self.queue.borrow_mut().pop_front() {
101                return Ok(output);
102            }
103
104            self.notify.notified().await;
105        }
106    }
107}
108
109#[cfg(test)]
110mod tests {
111    use std::sync::Arc;
112    use std::sync::atomic::{AtomicBool, Ordering};
113
114    use tokio::task;
115
116    use crate::Processor;
117
118    use super::{Hooks, ProcessorHook};
119
120    #[tokio::test]
121    async fn react_to_processor_events() {
122        type Event = String;
123
124        #[derive(Clone, Debug)]
125        struct BooHook {
126            result: Arc<AtomicBool>,
127        }
128
129        impl ProcessorHook<Event> for BooHook {
130            async fn on_input(&self, input: &Event) {
131                // Are you screaming already? .. if not, trigger hook!
132                self.result
133                    .swap(&input.to_uppercase() != input, Ordering::Relaxed);
134            }
135        }
136
137        #[derive(Clone, Debug)]
138        struct ZzzHook {
139            result: Arc<AtomicBool>,
140        }
141
142        impl ProcessorHook<Event> for ZzzHook {
143            async fn on_input(&self, input: &Event) {
144                // Are you sleeping already? .. if not, trigger hook!
145                self.result.swap(!input.is_empty(), Ordering::Relaxed);
146            }
147        }
148
149        let local = task::LocalSet::new();
150
151        local
152            .run_until(async {
153                let boo = BooHook {
154                    result: Arc::default(),
155                };
156
157                let zzz = ZzzHook {
158                    result: Arc::default(),
159                };
160
161                let mut processor = Hooks::<Event>::new();
162                processor.push(boo.clone());
163                processor.push(zzz.clone());
164
165                processor.process("I'm not scared.".into()).await.unwrap();
166                let _ = processor.next().await.unwrap();
167
168                assert!(boo.result.load(Ordering::Relaxed), "BOO, WE SCARED YOU!");
169                assert!(
170                    zzz.result.load(Ordering::Relaxed),
171                    "You should be sleeping!"
172                );
173
174                processor.process("AAAAAAAAAAH".into()).await.unwrap();
175                let _ = processor.next().await.unwrap();
176
177                assert!(!boo.result.load(Ordering::Relaxed), "AHAHAHAHA");
178                assert!(zzz.result.load(Ordering::Relaxed), "Try to sleep!");
179
180                processor.process("".into()).await.unwrap();
181                let _ = processor.next().await.unwrap();
182
183                assert!(!boo.result.load(Ordering::Relaxed), "Okay, you are silent.");
184                assert!(
185                    !zzz.result.load(Ordering::Relaxed),
186                    "Good, you are sleeping. Good night!"
187                );
188            })
189            .await;
190    }
191}
p2panda_stream/hooks.rs

p2panda_stream/
hooks.rs
