mermaid_cli/engine/
handle.rs

1//! The mailbox and the event bus of a running engine.
2//!
3//! [`Engine::drive`](super::Engine::drive) is already an actor loop: it pumps a
4//! channel of `Msg` and publishes through its observer. What was missing was a
5//! name for the two ends, so something outside the drive — a daemon socket, a
6//! second view of one session, an SDK client — can reach it. A handle is that
7//! name, and it is deliberately thin: both ends already existed, unnamed and
8//! reachable only by whoever happened to hold the raw channel.
9//!
10//! The mailbox is the SAME channel every effect result arrives on, and that is
11//! the point. A message sent from outside is indistinguishable from one the run
12//! produced itself, so it goes through the same reducer, the same stale-turn
13//! filter, the same recorder, and the same event log. There is no second way
14//! into the state.
15
16use std::fmt;
17
18use tokio::sync::{broadcast, mpsc};
19
20use mermaid_domain::Msg;
21
22/// The drive that owned this handle's inbox has ended.
23///
24/// Carries the message back, because the caller is usually the one place that
25/// can still do something with it — report it to a user, or persist it for the
26/// next run. Boxed: `Msg` is the large enum this codebase already carries an
27/// expect for, and an error that size rides every `Result` on the path.
28#[derive(Debug)]
29pub struct EngineGone(Box<Msg>);
30
31impl EngineGone {
32    /// The message that was not delivered.
33    #[must_use]
34    pub fn message(&self) -> &Msg {
35        &self.0
36    }
37
38    /// Take the undelivered message back.
39    #[must_use]
40    pub fn into_message(self) -> Msg {
41        *self.0
42    }
43}
44
45impl fmt::Display for EngineGone {
46    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
47        write!(f, "the engine is no longer running")
48    }
49}
50
51impl std::error::Error for EngineGone {}
52
53/// A running engine, reachable from outside its drive.
54///
55/// Cheap to clone: both halves are channel senders. Cloning does not keep the
56/// engine alive — the drive ends when its policy says so, and every handle then
57/// reports [`EngineGone`].
58pub struct EngineHandle<E> {
59    inbox: mpsc::Sender<Msg>,
60    events: broadcast::Sender<E>,
61}
62
63// Derived `Clone` would demand `E: Clone`, which is wrong: a `broadcast::Sender`
64// clones regardless of what it carries.
65impl<E> Clone for EngineHandle<E> {
66    fn clone(&self) -> Self {
67        Self {
68            inbox: self.inbox.clone(),
69            events: self.events.clone(),
70        }
71    }
72}
73
74impl<E> EngineHandle<E> {
75    /// Wrap an existing pair. The daemon supplies its own event bus, because it
76    /// has to subscribe *before* the run starts to catch the line that names
77    /// the session.
78    #[must_use]
79    pub const fn new(inbox: mpsc::Sender<Msg>, events: broadcast::Sender<E>) -> Self {
80        Self { inbox, events }
81    }
82
83    /// Wrap an inbox with a fresh event bus of `capacity`.
84    #[must_use]
85    pub fn with_capacity(inbox: mpsc::Sender<Msg>, capacity: usize) -> Self
86    where
87        E: Clone,
88    {
89        Self::new(inbox, broadcast::channel(capacity).0)
90    }
91
92    /// Deliver a message to the running engine.
93    ///
94    /// Awaits a slot when the inbox is full: that is backpressure from a run
95    /// that is behind on its own effects, and dropping the message instead
96    /// would lose a user's prompt. Use [`EngineHandle::try_send`] where the
97    /// caller cannot wait.
98    ///
99    /// # Errors
100    ///
101    /// [`EngineGone`] once the drive has ended.
102    pub async fn send(&self, msg: Msg) -> Result<(), EngineGone> {
103        self.inbox
104            .send(msg)
105            .await
106            .map_err(|e| EngineGone(Box::new(e.0)))
107    }
108
109    /// Deliver without waiting. Reports [`EngineGone`] for a finished engine
110    /// and, for a full inbox, hands the message back the same way — a caller
111    /// that cannot wait cannot queue either.
112    ///
113    /// # Errors
114    ///
115    /// [`EngineGone`] when the drive has ended or its inbox is full.
116    pub fn try_send(&self, msg: Msg) -> Result<(), EngineGone> {
117        use mpsc::error::TrySendError;
118        match self.inbox.try_send(msg) {
119            Ok(()) => Ok(()),
120            Err(TrySendError::Full(m) | TrySendError::Closed(m)) => Err(EngineGone(Box::new(m))),
121        }
122    }
123
124    /// Watch what the engine's observer publishes, from now on.
125    ///
126    /// A subscriber that needs what it missed reads the session event log first
127    /// and then joins here — the shape `subscribe_task` already uses, and the
128    /// reason this is a `broadcast` and not a replayable stream.
129    #[must_use]
130    pub fn subscribe(&self) -> broadcast::Receiver<E> {
131        self.events.subscribe()
132    }
133
134    /// The publishing end, for whatever observer feeds this handle.
135    #[must_use]
136    pub const fn publisher(&self) -> &broadcast::Sender<E> {
137        &self.events
138    }
139
140    /// Whether the drive is still pumping. Racy by nature — a run can end
141    /// between the check and the send, which is why [`EngineHandle::send`]
142    /// reports it too.
143    #[must_use]
144    pub fn is_running(&self) -> bool {
145        !self.inbox.is_closed()
146    }
147}
148
149impl<E> fmt::Debug for EngineHandle<E> {
150    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
151        f.debug_struct("EngineHandle")
152            .field("running", &self.is_running())
153            .field("subscribers", &self.events.receiver_count())
154            // What a reader wants is the state of the two ends, not the two
155            // senders themselves, which print as nothing useful.
156            .finish_non_exhaustive()
157    }
158}
mermaid_cli/engine/handle.rs

mermaid_cli/engine/
handle.rs
