botkit_cli/
transport.rs

1//! How events reach the bot and outbound actions reach drivers.
2//!
3//! `Stdio` reads `Inbound` lines from stdin and prints `Outbound` lines to
4//! stdout — the simplest mode for `cargo run` debugging.
5//!
6//! `Unix` binds a socket; each connection either streams inbound events
7//! (each answered with an `ack`/`error` line) or opens with
8//! `{"type":"subscribe"}` and streams every outbound line — the mode to use
9//! against a long-running bot process.
10//!
11//! `Manual` attaches no transport at all: the owner drives the bot through
12//! the [`CliHub`] handle directly (tests, in-process embedding).
13
14use std::io::{BufRead, BufReader, Write};
15use std::path::PathBuf;
16
17use futures_lite::io::{AsyncBufReadExt, AsyncWriteExt, BufReader as AsyncBufReader};
18
19use crate::hub::CliHub;
20use crate::wire::{Inbound, Outbound, OutboundAck, OutboundError};
21
22/// How the CLI platform moves events and actions.
23#[derive(Debug, Clone)]
24pub enum Transport {
25    /// Inbound events on stdin, outbound actions on stdout.
26    Stdio,
27    /// A unix socket at this path.
28    #[cfg(unix)]
29    Unix(PathBuf),
30    /// No transport — the owner injects events and reads outbound actions
31    /// through the [`CliHub`] handle.
32    Manual,
33}
34
35/// Start the transport: push inbound events into the hub and wire outbound
36/// sinks. Returns immediately; all work happens on spawned tasks/threads.
37pub(crate) fn start(transport: &Transport, hub: &CliHub) {
38    match transport {
39        Transport::Stdio => start_stdio(hub.clone()),
40        #[cfg(unix)]
41        Transport::Unix(path) => start_unix(path.clone(), hub.clone()),
42        Transport::Manual => {}
43    }
44}
45
46fn start_stdio(hub: CliHub) {
47    // stdout becomes the single implicit subscriber.
48    let lines = hub.subscribe();
49    executor_core::spawn(async move {
50        let stdout = std::io::stdout();
51        while let Ok(line) = lines.recv().await {
52            let mut out = stdout.lock();
53            let _ = writeln!(out, "{line}");
54            let _ = out.flush();
55        }
56    })
57    .detach();
58
59    // stdin is blocking IO; read it on a dedicated thread off the executor.
60    std::thread::spawn(move || {
61        let stdin = std::io::stdin();
62        for line in BufReader::new(stdin.lock()).lines() {
63            let Ok(line) = line else { break };
64            if line.trim().is_empty() {
65                continue;
66            }
67            match serde_json::from_str::<Inbound>(&line) {
68                Ok(event) => {
69                    let message_id = hub.inject(event);
70                    hub.emit(Outbound::Ack(OutboundAck {
71                        message_id: Some(message_id),
72                    }));
73                }
74                Err(error) => hub.emit(Outbound::Error(OutboundError {
75                    message: error.to_string(),
76                })),
77            }
78        }
79    });
80}
81
82#[cfg(unix)]
83fn start_unix(path: PathBuf, hub: CliHub) {
84    use async_io::Async;
85    use std::os::unix::net::UnixListener;
86
87    let _ = std::fs::remove_file(&path);
88    let listener = match UnixListener::bind(&path).and_then(Async::new) {
89        Ok(listener) => listener,
90        Err(error) => {
91            tracing::error!(%error, path = %path.display(), "cli transport failed to bind");
92            return;
93        }
94    };
95    tracing::info!(path = %path.display(), "cli transport listening");
96
97    executor_core::spawn(async move {
98        loop {
99            match listener.accept().await {
100                Ok((stream, _)) => {
101                    let hub = hub.clone();
102                    executor_core::spawn(handle_conn(stream, hub)).detach();
103                }
104                Err(error) => {
105                    tracing::warn!(%error, "cli transport accept failed");
106                }
107            }
108        }
109    })
110    .detach();
111}
112
113#[cfg(unix)]
114async fn handle_conn(stream: async_io::Async<std::os::unix::net::UnixStream>, hub: CliHub) {
115    use futures_lite::stream::StreamExt;
116    let (reader, mut writer) = futures_lite::io::split(stream);
117    let mut lines = AsyncBufReader::new(reader).lines();
118    // A driver connection is persistent: every event line gets an `ack` (or
119    // `error`) reply, so one connection can drive a whole session.
120    while let Some(line) = lines.next().await {
121        let Ok(line) = line else { return };
122        let reply = match serde_json::from_str::<Inbound>(&line) {
123            Ok(Inbound::Subscribe) => {
124                // Switch roles: this connection becomes an outbound sink.
125                let sink = hub.subscribe();
126                while let Ok(line) = sink.recv().await {
127                    if writer.write_all(line.as_bytes()).await.is_err()
128                        || writer.write_all(b"\n").await.is_err()
129                    {
130                        return;
131                    }
132                }
133                return;
134            }
135            Ok(event) => {
136                let message_id = hub.inject(event);
137                Outbound::Ack(OutboundAck {
138                    message_id: Some(message_id),
139                })
140            }
141            Err(error) => Outbound::Error(OutboundError {
142                message: error.to_string(),
143            }),
144        };
145        if let Ok(reply) = serde_json::to_string(&reply)
146            && (writer.write_all(reply.as_bytes()).await.is_err()
147                || writer.write_all(b"\n").await.is_err())
148        {
149            return;
150        }
151    }
152}
botkit_cli/transport.rs

botkit_cli/
transport.rs
