use std::io::Write as _; use std::net::SocketAddr; use tokio::net::TcpListener; use tokio_tungstenite::accept_async; use tracing::warn; use crate::ExecServerRuntimePaths; use crate::connection::JsonRpcConnection; use crate::server::processor::ConnectionProcessor; pub const DEFAULT_LISTEN_URL: &str = "ws://127.0.0.1:0"; #[derive(Debug, Clone, Eq, PartialEq)] pub enum ExecServerListenUrlParseError { UnsupportedListenUrl(String), InvalidWebSocketListenUrl(String), } impl std::fmt::Display for ExecServerListenUrlParseError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { ExecServerListenUrlParseError::UnsupportedListenUrl(listen_url) => write!( f, "unsupported --listen URL `{listen_url}`; expected `ws://IP:PORT`" ), ExecServerListenUrlParseError::InvalidWebSocketListenUrl(listen_url) => write!( f, "invalid websocket --listen URL `{listen_url}`; expected `ws://IP:PORT`" ), } } } impl std::error::Error for ExecServerListenUrlParseError {} pub(crate) fn parse_listen_url( listen_url: &str, ) -> Result { if let Some(socket_addr) = listen_url.strip_prefix("ws://") { return socket_addr.parse::().map_err(|_| { ExecServerListenUrlParseError::InvalidWebSocketListenUrl(listen_url.to_string()) }); } Err(ExecServerListenUrlParseError::UnsupportedListenUrl( listen_url.to_string(), )) } pub(crate) async fn run_transport( listen_url: &str, runtime_paths: ExecServerRuntimePaths, ) -> Result<(), Box> { let bind_address = parse_listen_url(listen_url)?; run_websocket_listener(bind_address, runtime_paths).await } async fn run_websocket_listener( bind_address: SocketAddr, runtime_paths: ExecServerRuntimePaths, ) -> Result<(), Box> { let listener = TcpListener::bind(bind_address).await?; let local_addr = listener.local_addr()?; let processor = ConnectionProcessor::new(runtime_paths); tracing::info!("codex-exec-server listening on ws://{local_addr}"); println!("ws://{local_addr}"); std::io::stdout().flush()?; loop { let (stream, peer_addr) = listener.accept().await?; let processor = processor.clone(); tokio::spawn(async move { match accept_async(stream).await { Ok(websocket) => { processor .run_connection(JsonRpcConnection::from_websocket( websocket, format!("exec-server websocket {peer_addr}"), )) .await; } Err(err) => { warn!( "failed to accept exec-server websocket connection from {peer_addr}: {err}" ); } } }); } } #[cfg(test)] #[path = "transport_tests.rs"] mod transport_tests;