From e00ceecab615c8a4dbea24883a93e31c8b079fa8 Mon Sep 17 00:00:00 2001 From: Serkyo Date: Wed, 22 Jul 2026 00:06:46 +0200 Subject: [PATCH] feat(net): expose the client chunk stream to the winit loop --- crates/net/src/chunk.rs | 77 +++++++++++++++++++++++++++++++++++++++ crates/net/src/lib.rs | 6 ++- crates/net/src/runtime.rs | 40 ++++++++++++++++---- 3 files changed, 113 insertions(+), 10 deletions(-) diff --git a/crates/net/src/chunk.rs b/crates/net/src/chunk.rs index 5df814e..a2598e0 100644 --- a/crates/net/src/chunk.rs +++ b/crates/net/src/chunk.rs @@ -94,6 +94,83 @@ pub(crate) async fn chunk_stream_task( } } +/// A synchronous handle the client's UI thread uses to push [`ChunkSubscribe`] requests to its network task. +/// +/// The client-side mirror of [`ChunkSink`]: `send` is synchronous and callable from the winit loop with no runtime in scope, while the client's chunk task drains the receiver with `recv().await` so it composes into that task's `select!`. +#[derive(Debug, Clone)] +pub struct ChunkSubscriber { + /// Outbound queue of subscription updates drained by the client's chunk task. + tx: UnboundedSender, +} + +impl ChunkSubscriber { + /// Wraps `tx` as a chunk subscriber. + pub(crate) fn new(tx: UnboundedSender) -> Self { + Self { tx } + } + + /// Queues a subscription update for the server. + /// + /// Non-blocking. A send failure means the network task has ended (the connection dropped); it is logged at debug and swallowed, since the UI thread cannot act on a departed connection. + pub fn send(&self, request: ChunkSubscribe) { + if self.tx.send(request).is_err() { + debug!("chunk subscriber send failed; network task has ended"); + } + } +} + +/// Runs the client-side chunk-stream pump for one connection until the stream or connection closes. +/// +/// Opens the chunk stream, then loops: [`ChunkSubscribe`] requests taken from `subscribe` are written to the server, and inbound [`ChunkMessage`] frames are forwarded to the UI thread over `deliveries`. The loop ends when the UI drops its subscriber, when the delivery receiver is gone, or when the stream closes. +pub(crate) async fn client_chunk_task( + connection: quinn::Connection, + mut subscribe: UnboundedReceiver, + deliveries: crossbeam_channel::Sender, +) { + // The client opens the chunk stream after the handshake; the server accepts it, mirroring the control-stream convention. + let (mut send, mut recv) = match connection.open_bi().await { + Ok(stream) => stream, + Err(error) => { + warn!(%error, "failed to open chunk stream"); + return; + } + }; + + loop { + tokio::select! { + // The UI thread pushed a subscription update to forward to the server. + request = subscribe.recv() => { + match request { + Some(request) => { + if let Err(error) = write_frame(&mut send, &request).await { + warn!(%error, "failed to write chunk subscribe; ending chunk stream"); + break; + } + } + // The subscriber was dropped: the UI is shutting down. + None => break, + } + } + // A chunk arrived from the server. + frame = read_frame::(&mut recv, MAX_CHUNK_FRAME_LEN) => { + match frame { + Ok(message) => { + // A closed delivery receiver means the UI is gone; nothing more to do. + if deliveries.send(message).is_err() { + break; + } + } + Err(error) => { + // A read error is the normal end of the session (stream finished or reset). + debug!(%error, "chunk stream read ended"); + break; + } + } + } + } + } +} + #[cfg(test)] #[path = "tests/chunk.rs"] mod tests; diff --git a/crates/net/src/lib.rs b/crates/net/src/lib.rs index 7d8e08c..ff81cb3 100644 --- a/crates/net/src/lib.rs +++ b/crates/net/src/lib.rs @@ -13,8 +13,10 @@ pub mod error; pub mod handshake; pub mod runtime; -pub use chunk::ChunkSink; -pub use runtime::{ConnectOutcome, NetworkServer, ServerEvent, connect_in_background}; +pub use chunk::{ChunkSink, ChunkSubscriber}; +pub use runtime::{ + ChunkStream, ClientLink, ConnectOutcome, NetworkServer, ServerEvent, connect_in_background, +}; /// Default UDP port the server binds and the client connects to when none is configured. // TODO: make the bind address and port configurable through server/client configuration. diff --git a/crates/net/src/runtime.rs b/crates/net/src/runtime.rs index 99b08ed..27b21c2 100644 --- a/crates/net/src/runtime.rs +++ b/crates/net/src/runtime.rs @@ -5,11 +5,11 @@ use std::net::SocketAddr; use std::thread; -use shared::protocol::chunk::ChunkSubscribe; +use shared::protocol::chunk::{ChunkMessage, ChunkSubscribe}; use shared::protocol::{ClientHello, HandshakeAck}; use tracing::{info, warn}; -use crate::chunk::{ChunkSink, chunk_stream_task}; +use crate::chunk::{ChunkSink, ChunkSubscriber, chunk_stream_task, client_chunk_task}; use crate::endpoint::{client_endpoint, server_endpoint}; use crate::error::NetError; use crate::handshake::{ServerConnection, accept_connection, connect}; @@ -17,6 +17,21 @@ use crate::handshake::{ServerConnection, accept_connection, connect}; /// Channel receiver delivering the outcome of a background client connect: the negotiated [`HandshakeAck`] on success, or a human-readable error string on failure. pub type ConnectOutcome = crossbeam_channel::Receiver>; +/// Non-blocking receiver of chunks delivered by the server, drained by the UI thread with `try_recv`. +pub type ChunkStream = crossbeam_channel::Receiver; + +/// Handles a background client connection exposes to the synchronous UI thread. +/// +/// The network task keeps the QUIC connection alive on its own thread; this bundle is how the winit loop observes the handshake outcome, pushes subscription updates, and drains chunk deliveries, all without touching the async runtime. +pub struct ClientLink { + /// Handshake outcome, drained once for the negotiated ack or the failure reason. + pub handshake: ConnectOutcome, + /// Sends subscription updates (center and radius) to the server as the camera moves. + pub subscribe: ChunkSubscriber, + /// Receives chunk deliveries from the server, drained non-blocking each frame. + pub chunks: ChunkStream, +} + /// An event surfaced by the network thread to the synchronous server loop. #[derive(Debug)] pub enum ServerEvent { @@ -230,12 +245,17 @@ async fn handle_connection( } } -/// Runs a one-shot connect and Synvael handshake against `server_addr` on a background `tokio` thread, reporting the outcome to the returned receiver. +/// Runs a connect and Synvael handshake against `server_addr` on a background `tokio` thread, then pumps the chunk stream, returning the handles the UI thread uses to observe and drive the connection. +/// +/// The returned [`ClientLink`] is available immediately; its channels buffer until the handshake completes and the chunk task starts. A handshake failure is reported on `handshake` and leaves the subscribe and chunk channels inert. #[must_use] -pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> ConnectOutcome { +pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> ClientLink { let (outcome_tx, outcome_rx) = crossbeam_channel::bounded(1); // Retained so a failure to spawn the thread can still be reported to the caller. let spawn_err_tx = outcome_tx.clone(); + // Subscription updates flow UI -> network (sync send, async recv); chunk deliveries flow network -> UI (async send, sync try_recv). + let (subscribe_tx, subscribe_rx) = tokio::sync::mpsc::unbounded_channel::(); + let (chunks_tx, chunks_rx) = crossbeam_channel::unbounded::(); let spawned = thread::Builder::new() .name("net-client".to_owned()) @@ -266,9 +286,9 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Con if outcome_tx.send(Ok(connected.ack.clone())).is_err() { return; } - // Keep the connection alive on the network thread until the server closes it. A full client session pump is a later concept. - let reason = connected.connection.closed().await; - info!(%reason, "server connection closed"); + // Pump the chunk stream on this thread until the UI drops its handles or the server closes the connection. + client_chunk_task(connected.connection, subscribe_rx, chunks_tx).await; + warn!("server connection closed"); } Err(error) => { let _ = outcome_tx.send(Err(error.to_string())); @@ -281,5 +301,9 @@ pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> Con let _ = spawn_err_tx.send(Err(format!("failed to spawn network thread: {error}"))); } - outcome_rx + ClientLink { + handshake: outcome_rx, + subscribe: ChunkSubscriber::new(subscribe_tx), + chunks: chunks_rx, + } }