sequoia_async/
emitter.rs

1#![forbid(unsafe_code)]
2
3use std::any::Any;
4use std::sync::mpsc::Sender;
5
6use crate::wake::WakeFn;
7
8#[derive(Clone, Copy, Debug, PartialEq, Eq)]
9pub struct StreamError;
10
11impl std::fmt::Display for StreamError {
12    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
13        write!(f, "async stream closed")
14    }
15}
16
17impl std::error::Error for StreamError {}
18
19#[doc(hidden)]
20pub struct StreamMsgAny {
21    pub id: u64,
22
23    pub kind: StreamKind,
24}
25
26#[doc(hidden)]
27#[non_exhaustive]
28pub enum StreamKind {
29    Item(Box<dyn Any + Send>),
30
31    Finished,
32
33    Panicked,
34}
35
36pub struct Emitter<T: Send + 'static> {
37    id: u64,
38    tx: Sender<StreamMsgAny>,
39    wake: WakeFn,
40    _marker: std::marker::PhantomData<T>,
41}
42
43impl<T: Send + 'static> Emitter<T> {
44    #[doc(hidden)]
45    pub fn new(id: u64, tx: Sender<StreamMsgAny>, wake: WakeFn) -> Self {
46        Self {
47            id,
48            tx,
49            wake,
50            _marker: std::marker::PhantomData,
51        }
52    }
53
54    pub fn id(&self) -> u64 {
55        self.id
56    }
57
58    pub fn emit(&self, value: T) -> Result<(), StreamError> {
59        self.tx
60            .send(StreamMsgAny {
61                id: self.id,
62                kind: StreamKind::Item(Box::new(value)),
63            })
64            .map_err(|_| StreamError)?;
65        (self.wake)();
66        Ok(())
67    }
68}
69
70impl<T: Send + 'static> Clone for Emitter<T> {
71    fn clone(&self) -> Self {
72        Self {
73            id: self.id,
74            tx: self.tx.clone(),
75            wake: self.wake.clone(),
76            _marker: std::marker::PhantomData,
77        }
78    }
79}
80
81impl<T: Send + 'static> std::fmt::Debug for Emitter<T> {
82    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83        f.debug_struct("Emitter").field("id", &self.id).finish()
84    }
85}
86
87#[cfg(test)]
88mod tests {
89    use super::*;
90    use crate::wake::noop_wake;
91    use std::sync::mpsc::channel;
92
93    #[test]
94    fn emitter_sends_item_and_wakes() {
95        use std::sync::Arc;
96        use std::sync::atomic::{AtomicUsize, Ordering};
97        let (tx, rx) = channel::<StreamMsgAny>();
98        let n = Arc::new(AtomicUsize::new(0));
99        let probe = Arc::clone(&n);
100        let wake = crate::wake::fn_wake(move || {
101            probe.fetch_add(1, Ordering::SeqCst);
102        });
103        let em: Emitter<u32> = Emitter::new(9, tx, wake);
104        assert!(em.emit(42).is_ok());
105        assert_eq!(n.load(Ordering::SeqCst), 1);
106        let msg = rx.try_recv().expect("item queued");
107        assert_eq!(msg.id, 9);
108        match msg.kind {
109            StreamKind::Item(any) => assert_eq!(*any.downcast::<u32>().unwrap(), 42),
110            _ => panic!("expected item"),
111        }
112        let _ = noop_wake();
113    }
114
115    #[test]
116    fn emitter_errors_after_receiver_dropped() {
117        let (tx, rx) = channel::<StreamMsgAny>();
118        drop(rx);
119        let em: Emitter<u32> = Emitter::new(1, tx, noop_wake());
120        assert_eq!(em.emit(1).unwrap_err(), StreamError);
121    }
122}
sequoia_async/emitter.rs

sequoia_async/
emitter.rs
