// TcpBridge - TCP localhost IPC implementation // Implements D-020 subprocess/IPC architecture // Deterministic client-server communication via TCP sockets // Used for Godot client which lacks Unix socket support use super::{BridgeError, ObserverSnapshot, PlayerInput, SimBridge}; use crate::bridge::framing::{read_framed, write_framed}; use std::io::{BufReader, BufWriter}; use std::net::{SocketAddr, TcpListener, TcpStream}; use std::sync::Mutex; /// TcpBridge: TCP transport for client-server IPC /// Same semantics as LocalBridge but over TCP localhost pub struct TcpBridge { reader: Mutex>, writer: Mutex>, local_addr: SocketAddr, } impl TcpBridge { /// Server-side: bind TCP listener and accept one connection. /// Binds to the specified address (e.g., "127.0.0.1:0" for OS-assigned port). /// Returns the bridge with the actual bound address available via local_addr(). pub fn accept(addr: &str) -> Result { tracing::info!("TcpBridge binding to {}", addr); let listener = TcpListener::bind(addr) .map_err(|e| BridgeError::Transport(format!("failed to bind TCP socket: {}", e)))?; let local_addr = listener .local_addr() .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?; tracing::info!("TcpBridge listening on {}", local_addr); // Accept one connection let (stream, peer_addr) = listener .accept() .map_err(|e| BridgeError::Transport(format!("failed to accept connection: {}", e)))?; tracing::info!( "TcpBridge accepted connection from {} on {}", peer_addr, local_addr ); // Set non-blocking so receive_inputs doesn't stall the game loop. // read_framed handles WouldBlock by returning Ok(None). stream .set_nonblocking(true) .map_err(|e| BridgeError::Transport(format!("failed to set non-blocking: {}", e)))?; // Clone stream for reader and writer let reader_stream = stream.try_clone().map_err(|e| { BridgeError::Transport(format!("failed to clone stream for reader: {}", e)) })?; Ok(Self { reader: Mutex::new(BufReader::new(reader_stream)), writer: Mutex::new(BufWriter::new(stream)), local_addr, }) } /// Server-side: accept one connection on an existing TcpListener. /// Avoids race conditions in tests by separating bind from accept. pub fn accept_on(listener: TcpListener) -> Result { let local_addr = listener .local_addr() .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?; tracing::info!("TcpBridge accepting on {}", local_addr); let (stream, peer_addr) = listener .accept() .map_err(|e| BridgeError::Transport(format!("failed to accept connection: {}", e)))?; tracing::info!( "TcpBridge accepted connection from {} on {}", peer_addr, local_addr ); stream .set_nonblocking(true) .map_err(|e| BridgeError::Transport(format!("failed to set non-blocking: {}", e)))?; let reader_stream = stream.try_clone().map_err(|e| { BridgeError::Transport(format!("failed to clone stream for reader: {}", e)) })?; Ok(Self { reader: Mutex::new(BufReader::new(reader_stream)), writer: Mutex::new(BufWriter::new(stream)), local_addr, }) } /// Client-side: connect to TCP address (for tests). pub fn connect(addr: &str) -> Result { tracing::info!("TcpBridge connecting to {}", addr); let stream = TcpStream::connect(addr).map_err(|e| { BridgeError::Transport(format!("failed to connect to TCP socket: {}", e)) })?; let local_addr = stream .local_addr() .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?; tracing::trace!("TcpBridge connected to {}", addr); // Clone stream for reader and writer let reader_stream = stream.try_clone().map_err(|e| { BridgeError::Transport(format!("failed to clone stream for reader: {}", e)) })?; Ok(Self { reader: Mutex::new(BufReader::new(reader_stream)), writer: Mutex::new(BufWriter::new(stream)), local_addr, }) } /// Get the local address (useful for OS-assigned port discovery in tests). pub fn local_addr(&self) -> SocketAddr { self.local_addr } } impl SimBridge for TcpBridge { fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError> { let payload = rmp_serde::to_vec_named(snapshot)?; let mut writer = self .writer .lock() .map_err(|e| BridgeError::MutexPoisoned(format!("writer: {}", e)))?; // Toggle to blocking for reliable write delivery. // Single-threaded bevy guarantees no concurrent reads during this window. let stream = writer.get_mut(); stream.set_nonblocking(false).map_err(BridgeError::Io)?; let result = write_framed(stream, &payload); stream.set_nonblocking(true).map_err(BridgeError::Io)?; result?; tracing::trace!("sent snapshot: tick={}", snapshot.tick); Ok(()) } fn receive_inputs(&self) -> Result, BridgeError> { let mut reader = self .reader .lock() .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?; match read_framed(reader.get_mut()) { Ok(Some(payload)) => match rmp_serde::from_slice::>(&payload) { Ok(inputs) => { tracing::trace!("received {} inputs", inputs.len()); Ok(inputs) } Err(e) => { let dump_len = payload.len().min(256); tracing::error!( "deserialization failed: {}. Raw bytes ({} of {} total): {:02x?}", e, dump_len, payload.len(), &payload[..dump_len] ); Err(BridgeError::DeserializationWithDump(format!( "{} (payload {} bytes)", e, payload.len() ))) } }, Ok(None) => Err(BridgeError::Disconnected), // Non-blocking socket: no data available this tick — not an error. Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(vec![]), Err(e) => Err(BridgeError::Io(e)), } } }