use std::io::Write;
#[derive(Clone)]
pub(crate) struct RpcWriter {
sender: tokio::sync::mpsc::Sender<Vec<u8>>,
mode: crate::args::RpcMessageUpdates,
}
pub(crate) fn start_rpc_writer<W: Write + Send + 'static>(
mut output: W,
mode: crate::args::RpcMessageUpdates,
) -> (
RpcWriter,
tokio::task::JoinHandle<std::io::Result<()>>,
tokio_util::sync::CancellationToken,
) {
let (tx, mut rx) = tokio::sync::mpsc::channel::<Vec<u8>>(64);
let failed = tokio_util::sync::CancellationToken::new();
let failure_signal = failed.clone();
let task = tokio::task::spawn_blocking(move || {
let result = (|| {
while let Some(line) = rx.blocking_recv() {
output.write_all(&line)?;
output.flush()?;
}
output.flush()
})();
if result.is_err() {
failure_signal.cancel();
}
result
});
(RpcWriter { sender: tx, mode }, task, failed)
}
pub(crate) async fn write_line(out: &RpcWriter, value: &serde_json::Value) {
let mut line = value.to_string().into_bytes();
line.push(b'\n');
let _ = out.sender.send(line).await;
}
pub(crate) async fn write_agent_event(out: &RpcWriter, event: &rpi_agent::events::AgentEvent) {
let mut line = super::events::encode(event, out.mode).unwrap_or_else(|_| b"null".to_vec());
line.push(b'\n');
let _ = out.sender.send(line).await;
}