syncweb_core/bridge/
server.rs

1use std::{net::SocketAddr, path::PathBuf, sync::Arc};
2
3use n0_future::split;
4use tokio::sync::broadcast;
5
6use crate::{
7    error::{Result, SyncwebError},
8    node::iroh_node::IrohNode,
9};
10
11use super::{service::BridgeService, session::WsSession};
12
13/// WebSocket bridge server that exposes Iroh P2P primitives over WebSocket.
14pub struct WsBridgeServer {
15    node: Arc<IrohNode>,
16    addr: SocketAddr,
17    data_dir: PathBuf,
18}
19
20impl WsBridgeServer {
21    #[must_use]
22    pub const fn new(node: Arc<IrohNode>, addr: SocketAddr, data_dir: PathBuf) -> Self {
23        Self { node, addr, data_dir }
24    }
25
26    /// Run the bridge server, accepting connections until shutdown is
27    /// signalled.
28    ///
29    /// # Errors
30    ///
31    /// Returns an error if the TCP listener cannot be bound or if an
32    /// accepted connection produces a fatal error.
33    pub async fn run(self, shutdown: broadcast::Sender<()>) -> Result<()> {
34        let listener = Self::bind_listener(self.addr).await?;
35
36        tracing::info!("WebSocket bridge listening on {}", self.addr);
37
38        let service = Arc::new(BridgeService::new(self.node, self.data_dir, shutdown.clone())?);
39
40        let mut server_shutdown = shutdown.subscribe();
41
42        loop {
43            tokio::select! {
44                _ = server_shutdown.recv() => {
45                    tracing::info!("bridge server shutting down");
46                    break Ok(());
47                }
48                accepted = listener.accept() => {
49                    Self::handle_accept(accepted, &service, &shutdown);
50                }
51            }
52        }
53    }
54
55    async fn bind_listener(addr: SocketAddr) -> Result<tokio::net::TcpListener> {
56        tokio::net::TcpListener::bind(addr)
57            .await
58            .map_err(|error| SyncwebError::operation(format!("failed to bind bridge server on {addr}"), error))
59    }
60
61    fn handle_accept(
62        accepted: std::io::Result<(tokio::net::TcpStream, SocketAddr)>,
63        service: &Arc<BridgeService>,
64        shutdown: &broadcast::Sender<()>,
65    ) {
66        let (stream, peer_addr) = match accepted {
67            Ok(conn) => conn,
68            Err(error) => {
69                tracing::warn!(%error, "bridge accept failed");
70                return;
71            }
72        };
73        let svc = Arc::clone(service);
74        let shutdown_rx = shutdown.subscribe();
75        tokio::spawn(async move {
76            if let Err(error) = handle_connection(stream, svc, shutdown_rx).await {
77                tracing::warn!(%peer_addr, %error, "bridge session error");
78            }
79        });
80    }
81}
82
83async fn handle_connection(
84    stream: tokio::net::TcpStream,
85    service: Arc<BridgeService>,
86    shutdown: broadcast::Receiver<()>,
87) -> Result<()> {
88    let ws_stream = tokio_tungstenite::accept_async(stream)
89        .await
90        .map_err(|error| SyncwebError::operation("websocket handshake failed", error))?;
91
92    let (sender, mut receiver) = split::split(ws_stream);
93
94    let session = WsSession::new(service, sender, shutdown);
95    session.run(&mut receiver).await
96}
syncweb_core/bridge/server.rs

syncweb_core/bridge/
server.rs
