syncweb_core/media/
bridge.rs

1use std::net::SocketAddr;
2
3use futures_util::{SinkExt, StreamExt};
4use tokio::sync::broadcast;
5use tokio::task::JoinHandle;
6use tokio_tungstenite::tungstenite::Message;
7
8use crate::SyncwebError;
9use crate::daemon::{IpcRequest, IpcResponse, IpcServer};
10
11/// Default listen address for the local WebSocket bridge.
12pub const DEFAULT_BRIDGE_LISTEN: &str = "127.0.0.1:9192";
13
14/// A local WebSocket bridge that proxies JSON [`IpcRequest`] messages to the
15/// daemon's control channel and streams the JSON [`IpcResponse`] back.
16///
17/// This lets scriptable clients (browsers, `websocat`, monitoring tools) drive
18/// the daemon over the same surface the CLI uses over its Unix socket.
19pub struct BridgeServer {
20    addr: SocketAddr,
21    ipc: IpcServer,
22}
23
24impl BridgeServer {
25    /// Construct a bridge from a listen address string.
26    ///
27    /// # Errors
28    ///
29    /// Returns an error if the listen address string cannot be parsed.
30    pub fn new(listen_addr: &str, ipc: IpcServer) -> Result<Self, SyncwebError> {
31        let addr = listen_addr
32            .parse::<SocketAddr>()
33            .map_err(|error| SyncwebError::operation("invalid websocket bridge listen address", error))?;
34        Ok(Self { addr, ipc })
35    }
36
37    /// Bind the TCP listener and run until the shutdown signal is received.
38    ///
39    /// # Errors
40    ///
41    /// Returns an error if the TCP listener cannot be bound.
42    pub async fn run(self, shutdown: broadcast::Sender<()>) -> Result<(), SyncwebError> {
43        let listener = tokio::net::TcpListener::bind(self.addr)
44            .await
45            .map_err(|error| SyncwebError::operation("failed to bind websocket bridge", error))?;
46        let local_addr = listener
47            .local_addr()
48            .map_err(|error| SyncwebError::operation("failed to get websocket bridge local address", error))?;
49        tracing::info!(addr = %local_addr, "websocket bridge started");
50
51        let mut shutdown_rx = shutdown.subscribe();
52        loop {
53            tokio::select! {
54                shutdown_result = shutdown_rx.recv() => {
55                    match shutdown_result {
56                        Ok(()) | Err(broadcast::error::RecvError::Closed) => break,
57                        Err(broadcast::error::RecvError::Lagged(_)) => {}
58                    }
59                }
60                accepted = listener.accept() => {
61                    let (stream, peer) = accepted.map_err(|error| {
62                        SyncwebError::operation("websocket bridge accept failed", error)
63                    })?;
64                    let ipc = self.ipc.clone();
65                    tokio::spawn(async move {
66                        if let Err(error) = Box::pin(serve_connection(stream, peer, ipc)).await {
67                            tracing::warn!(%peer, %error, "websocket bridge connection failed");
68                        }
69                    });
70                }
71            }
72        }
73        tracing::debug!("websocket bridge shutdown signal received");
74        Ok(())
75    }
76
77    #[must_use]
78    pub fn spawn(self, shutdown: broadcast::Sender<()>) -> JoinHandle<Result<(), SyncwebError>> {
79        tokio::spawn(async move { self.run(shutdown).await })
80    }
81}
82
83async fn serve_connection(stream: tokio::net::TcpStream, peer: SocketAddr, ipc: IpcServer) -> Result<(), SyncwebError> {
84    let websocket = tokio_tungstenite::accept_async(stream)
85        .await
86        .map_err(|error| SyncwebError::operation("websocket bridge handshake failed", error))?;
87    tracing::debug!(%peer, "websocket bridge connection accepted");
88    let (mut sink, mut incoming) = websocket.split();
89    while let Some(item) = incoming.next().await {
90        let message = item.map_err(|error| SyncwebError::operation("websocket bridge receive failed", error))?;
91        match message {
92            Message::Text(text) => {
93                let response = match serde_json::from_str::<IpcRequest>(text.as_ref()) {
94                    Ok(request) => ipc.handle_request(request).await,
95                    Err(error) => IpcResponse::Error {
96                        message: format!("invalid JSON: {error}"),
97                    },
98                };
99                let payload = serde_json::to_string(&response)
100                    .map_err(|error| SyncwebError::operation("failed to serialize bridge response", error))?;
101                sink.send(Message::Text(payload.into()))
102                    .await
103                    .map_err(|error| SyncwebError::operation("websocket bridge send failed", error))?;
104            }
105            Message::Close(_) => break,
106            Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {}
107        }
108    }
109    Ok(())
110}
syncweb_core/media/bridge.rs

syncweb_core/media/
bridge.rs
