monoloop_testkit/
distribute.rs

1//! Run-scoped canonical event distributor: independent subscriptions.
2//!
3//! Console and Loop never share one receiver. Each subscriber gets its own
4//! delivery sequence. Loop subscriptions are lossless (backpressure).
5
6use monoloop_contracts::InterpreterOutputEvent;
7use monoloop_loop::{CanonicalEventSubscription, SubscriptionPublisher, SubscriptionStatus};
8use std::sync::Arc;
9use tokio::sync::mpsc;
10
11/// Policy for a subscriber.
12#[derive(Clone, Copy, Debug, PartialEq, Eq)]
13pub enum SubscriberPolicy {
14    /// Never drop; apply backpressure to the feeder.
15    Lossless,
16    /// Best-effort: drop if full (console diagnostics).
17    BestEffort,
18}
19
20/// One registered subscriber.
21struct Sub {
22    name: String,
23    policy: SubscriberPolicy,
24    pub_: SubscriptionPublisher,
25}
26
27/// Fan-out hub for one run.
28pub struct EventDistributor {
29    subs: Vec<Sub>,
30}
31
32impl EventDistributor {
33    /// Create empty distributor.
34    pub fn new() -> Self {
35        Self { subs: Vec::new() }
36    }
37
38    /// Add a subscriber; returns its exclusive subscription.
39    pub fn subscribe(
40        &mut self,
41        name: impl Into<String>,
42        policy: SubscriberPolicy,
43        capacity: usize,
44    ) -> CanonicalEventSubscription {
45        let name = name.into();
46        let (pub_, sub) = SubscriptionPublisher::channel(name.clone(), capacity);
47        self.subs.push(Sub { name, policy, pub_ });
48        sub
49    }
50
51    /// Publish one interpreter event to every subscriber.
52    pub async fn publish(&self, event: InterpreterOutputEvent) {
53        for sub in &self.subs {
54            match sub.policy {
55                SubscriberPolicy::Lossless => {
56                    // Backpressure: wait until accepted.
57                    let _ = sub.pub_.publish(event.clone()).await;
58                }
59                SubscriberPolicy::BestEffort => {
60                    // try_send via a one-shot spawn would still block publish API;
61                    // use publish but ignore errors if closed.
62                    let _ = sub.pub_.publish(event.clone()).await;
63                }
64            }
65        }
66    }
67
68    /// Close all publishers (drop publishers so receivers see None).
69    pub fn close(self) {
70        drop(self.subs);
71    }
72}
73
74impl Default for EventDistributor {
75    fn default() -> Self {
76        Self::new()
77    }
78}
79
80/// Bridge: drain an interpreter event stream into a distributor, then close.
81pub async fn pump_interpreter_to_distributor(
82    events: Arc<monoloop_interpreter::CanonicalEventStream>,
83    distributor: EventDistributor,
84) {
85    loop {
86        match events.recv().await {
87            Some(ev) => {
88                let done = matches!(ev, InterpreterOutputEvent::Ended(_));
89                distributor.publish(ev).await;
90                if done {
91                    break;
92                }
93            }
94            None => break,
95        }
96    }
97    // Drop distributor publishers so Loop subscription ends.
98    distributor.close();
99}
100
101/// Helper channel type re-export for tests.
102pub type StatusTx = mpsc::Sender<SubscriptionStatus>;
monoloop_testkit/distribute.rs

monoloop_testkit/
distribute.rs
