Synvael/synvael
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

feat(net): expose the client chunk stream to the winit loop

Browse source
This commit is contained in:
Serkyo 2026-07-22 00:06:46 +02:00
parent 9dc7bba3a9
commit e00ceecab6
3 changed files with 113 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

77
crates/net/src/chunk.rs
View file

@ -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<ChunkSubscribe>,
}
impl ChunkSubscriber {
/// Wraps `tx` as a chunk subscriber.
pub(crate) fn new(tx: UnboundedSender<ChunkSubscribe>) -> 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<ChunkSubscribe>,
deliveries: crossbeam_channel::Sender<ChunkMessage>,
) {
// 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::<ChunkMessage>(&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;

6
crates/net/src/lib.rs
View file

@ -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.

40
crates/net/src/runtime.rs
View file

@ -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<Result<HandshakeAck, String>>;
/// Non-blocking receiver of chunks delivered by the server, drained by the UI thread with `try_recv`.
pub type ChunkStream = crossbeam_channel::Receiver<ChunkMessage>;
/// 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::<ChunkSubscribe>();
let (chunks_tx, chunks_rx) = crossbeam_channel::unbounded::<ChunkMessage>();
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,
}
}