monoloop_interpreter/
stream.rs

1//! Bounded canonical event stream (single Interpretation output).
2
3use monoloop_contracts::{InterpreterError, InterpreterOutputEvent};
4use std::sync::Arc;
5use tokio::sync::{mpsc, Mutex};
6
7/// Cloneable receiver handle for canonical events from one Interpretation.
8#[derive(Debug)]
9pub struct CanonicalEventStream {
10    rx: Mutex<mpsc::Receiver<InterpreterOutputEvent>>,
11}
12
13impl CanonicalEventStream {
14    pub(crate) fn new(rx: mpsc::Receiver<InterpreterOutputEvent>) -> Self {
15        Self { rx: Mutex::new(rx) }
16    }
17
18    /// Receive the next event. `None` when the stream is closed after terminal end
19    /// was already delivered (or the owner dropped without end — treated as loss).
20    pub async fn recv(&self) -> Option<InterpreterOutputEvent> {
21        let mut guard = self.rx.lock().await;
22        guard.recv().await
23    }
24}
25
26/// Shared publisher used by the interpretation owner task.
27#[derive(Clone)]
28pub(crate) struct EventPublisher {
29    tx: mpsc::Sender<InterpreterOutputEvent>,
30    count: Arc<std::sync::atomic::AtomicU64>,
31}
32
33impl EventPublisher {
34    pub(crate) fn new(capacity: usize) -> (Self, CanonicalEventStream) {
35        let (tx, rx) = mpsc::channel(capacity.max(1));
36        (
37            Self {
38                tx,
39                count: Arc::new(std::sync::atomic::AtomicU64::new(0)),
40            },
41            CanonicalEventStream::new(rx),
42        )
43    }
44
45    pub(crate) fn count(&self) -> u64 {
46        self.count.load(std::sync::atomic::Ordering::Relaxed)
47    }
48
49    /// Publish with backpressure. Does not drop events.
50    pub(crate) async fn publish(
51        &self,
52        event: InterpreterOutputEvent,
53    ) -> Result<(), InterpreterError> {
54        self.tx
55            .send(event)
56            .await
57            .map_err(|_| InterpreterError::backpressure())?;
58        self.count
59            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
60        Ok(())
61    }
62}
monoloop_interpreter/stream.rs

monoloop_interpreter/
stream.rs
