botkit_cli/
hub.rs

1//! Shared state between the transport, the dispatcher, and platform senders.
2//!
3//! The hub owns three things: the inbound queue transports push events onto,
4//! the set of outbound sinks (subscribed socket connections, or stdout in
5//! stdio mode), and the message-id counter for both directions.
6
7use std::sync::atomic::{AtomicI64, Ordering};
8use std::sync::{Arc, Mutex};
9
10use async_channel::{Receiver, Sender};
11
12use crate::wire::{Inbound, Outbound};
13
14/// Shared CLI-platform state.
15///
16/// Clone it freely: the dispatcher, `CliActionSender`s, acpbot-style
17/// platform senders, and transports all hold their own handle.
18#[derive(Clone)]
19pub struct CliHub {
20    inner: Arc<HubInner>,
21}
22
23struct HubInner {
24    inbound: Sender<Inbound>,
25    sinks: Mutex<Vec<Sender<String>>>,
26    next_message_id: AtomicI64,
27}
28
29impl CliHub {
30    /// Create a hub plus the receiver the dispatcher drains.
31    pub(crate) fn new() -> (Self, Receiver<Inbound>) {
32        let (inbound_tx, inbound_rx) = async_channel::unbounded();
33        let hub = Self {
34            inner: Arc::new(HubInner {
35                inbound: inbound_tx,
36                sinks: Mutex::new(Vec::new()),
37                next_message_id: AtomicI64::new(1),
38            }),
39        };
40        (hub, inbound_rx)
41    }
42
43    /// Assign the next platform message id.
44    pub fn next_message_id(&self) -> i64 {
45        self.inner.next_message_id.fetch_add(1, Ordering::Relaxed)
46    }
47
48    /// Queue an inbound event for dispatch, assigning a message id where the
49    /// driver omitted one. Returns the id the event carries.
50    ///
51    /// `Inbound::Subscribe` never reaches this function — transports turn it
52    /// into a sink registration instead.
53    pub(crate) fn inject(&self, mut event: Inbound) -> i64 {
54        event.ensure_message_id(|| self.next_message_id());
55        let id = event.message_id().unwrap_or(0);
56        if self.inner.inbound.try_send(event).is_err() {
57            self.emit(Outbound::Error(crate::wire::OutboundError {
58                message: "dispatcher is gone; event dropped".to_string(),
59            }));
60        }
61        id
62    }
63
64    /// Register a new outbound sink. The returned receiver yields one JSONL
65    /// line per outbound action until unsubscribed (channel dropped).
66    pub fn subscribe(&self) -> Receiver<String> {
67        let (tx, rx) = async_channel::unbounded();
68        self.inner.sinks.lock().expect("sinks poisoned").push(tx);
69        rx
70    }
71
72    /// Serialize an outbound action and fan it out to every sink. Dead sinks
73    /// are pruned.
74    pub fn emit(&self, outbound: Outbound) {
75        let Ok(line) = serde_json::to_string(&outbound) else {
76            return;
77        };
78        self.emit_line(&line);
79    }
80
81    /// Push a pre-serialized line to every sink.
82    pub fn emit_line(&self, line: &str) {
83        let mut sinks = self.inner.sinks.lock().expect("sinks poisoned");
84        sinks.retain(|sink| sink.try_send(line.to_string()).is_ok());
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91    use crate::wire::{InboundMessage, OutboundMessage, WireUser};
92
93    fn event() -> Inbound {
94        Inbound::Message(Box::new(InboundMessage {
95            chat: "c".to_string(),
96            user: WireUser {
97                id: "u".to_string(),
98                name: "n".to_string(),
99            },
100            message_id: None,
101            text: Some("hi".to_string()),
102            caption: None,
103            thread_id: None,
104            reply_to: None,
105            files: vec![],
106            sticker: None,
107            ambient: false,
108        }))
109    }
110
111    #[test]
112    fn inject_assigns_message_ids() {
113        let (hub, rx) = CliHub::new();
114        let first = hub.inject(event());
115        let second = hub.inject(event());
116        assert_eq!((first, second), (1, 2));
117        assert_eq!(rx.try_recv().unwrap().message_id(), Some(1));
118        assert_eq!(rx.try_recv().unwrap().message_id(), Some(2));
119    }
120
121    #[test]
122    fn emit_reaches_subscribers_as_jsonl() {
123        let (hub, _rx) = CliHub::new();
124        let sink = hub.subscribe();
125        hub.emit(Outbound::Message(OutboundMessage {
126            chat: "c".to_string(),
127            message_id: 1,
128            text: "hello".to_string(),
129            buttons: vec![],
130            reply_to: None,
131            thread_id: None,
132            extras: None,
133        }));
134        let line = sink.try_recv().unwrap();
135        let parsed: Outbound = serde_json::from_str(&line).unwrap();
136        assert!(matches!(parsed, Outbound::Message(_)));
137    }
138
139    #[test]
140    fn dead_sinks_are_pruned() {
141        let (hub, _rx) = CliHub::new();
142        let sink = hub.subscribe();
143        drop(sink);
144        hub.emit(Outbound::Ack(crate::wire::OutboundAck { message_id: None }));
145        assert!(hub.inner.sinks.lock().unwrap().is_empty());
146    }
147}
botkit_cli/hub.rs

botkit_cli/
hub.rs
