Sasedev/games
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.3.4-alpha.4

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-21 18:00:32 +02:00
parent bbecae5e0d
commit fa17ea0ca5
12 changed files with 792 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
crates/common/game-realtime-websocket-lib/src/websocket.rs
View File

@@ -1,5 +1,5 @@
// file: crates/common/game-realtime-websocket-lib/src/websocket.rs
// version: 1
// version: 2
use futures_util::SinkExt; // rust-rules: trait-import
use futures_util::StreamExt; // rust-rules: trait-import
@@ -31,15 +31,16 @@ enum WebSocketReceiverKind {
/// Established WebSocket connection implementing the transport-neutral realtime contract.
pub struct WebSocketConnection {
inner: WebSocketStreamKind,
config: crate::WebSocketConfig,
}
impl WebSocketConnection {
fn from_client(stream: ClientStream) -> Self {
return Self { inner: WebSocketStreamKind::Client(stream) };
fn from_client(stream: ClientStream, config: crate::WebSocketConfig) -> Self {
return Self { inner: WebSocketStreamKind::Client(stream), config };
}
fn from_server(stream: ServerStream) -> Self {
return Self { inner: WebSocketStreamKind::Server(stream) };
fn from_server(stream: ServerStream, config: crate::WebSocketConfig) -> Self {
return Self { inner: WebSocketStreamKind::Server(stream), config };
}
}
@@ -48,18 +49,36 @@ impl game_realtime_transport_lib::RealtimeConnection for WebSocketConnection {
type Receiver = crate::WebSocketReceiver;
fn split(self) -> (Self::Sender, Self::Receiver) {
let max_frame_size = self.config.max_frame_size();
let max_message_size = self.config.max_message_size();
let send_timeout = self.config.send_timeout();
let close_timeout = self.config.close_timeout();
return match self.inner {
WebSocketStreamKind::Client(stream) => {
let (sender, receiver) = stream.split();
(
crate::WebSocketSender { inner: WebSocketSinkKind::Client(sender) },
crate::WebSocketSender {
inner: WebSocketSinkKind::Client(sender),
max_frame_size,
max_message_size,
send_timeout,
close_timeout,
send_timed_out: false,
},
crate::WebSocketReceiver { inner: WebSocketReceiverKind::Client(receiver) },
)
},
WebSocketStreamKind::Server(stream) => {
let (sender, receiver) = stream.split();
(
crate::WebSocketSender { inner: WebSocketSinkKind::Server(sender) },
crate::WebSocketSender {
inner: WebSocketSinkKind::Server(sender),
max_frame_size,
max_message_size,
send_timeout,
close_timeout,
send_timed_out: false,
},
crate::WebSocketReceiver { inner: WebSocketReceiverKind::Server(receiver) },
)
},
@@ -70,6 +89,11 @@ impl game_realtime_transport_lib::RealtimeConnection for WebSocketConnection {
/// Send half of an established WebSocket transport connection.
pub struct WebSocketSender {
inner: WebSocketSinkKind,
max_frame_size: usize,
max_message_size: usize,
send_timeout: std::time::Duration,
close_timeout: std::time::Duration,
send_timed_out: bool,
}
impl game_realtime_transport_lib::RealtimeSender for WebSocketSender {
@@ -84,42 +108,81 @@ impl game_realtime_transport_lib::RealtimeSender for WebSocketSender {
fn send(&mut self, message: game_realtime_transport_lib::TransportMessage) -> Self::SendFuture<'_> {
return Box::pin(async move {
if self.send_timed_out {
return Err(game_realtime_transport_lib::TransportError::new(
game_realtime_transport_lib::TransportErrorKind::Aborted,
"sender is unavailable after a previous send timeout",
));
}
let payload_len = message.len();
if payload_len > self.max_message_size || payload_len > self.max_frame_size {
let error = game_realtime_transport_lib::TransportError::new(
game_realtime_transport_lib::TransportErrorKind::MessageTooLarge,
format!("binary payload size {payload_len} exceeds configured message/frame maxima {}/{}", self.max_message_size, self.max_frame_size),
);
tracing::warn!(
target: TRACING_TARGET,
payload_len = payload_len,
max_message_size = self.max_message_size,
max_frame_size = self.max_frame_size,
"outbound WebSocket payload rejected"
);
return Err(error);
}
let websocket_message = tokio_tungstenite::tungstenite::Message::Binary(message.into_bytes().into());
let result = match &mut self.inner {
WebSocketSinkKind::Client(sender) => sender.send(websocket_message).await,
WebSocketSinkKind::Server(sender) => sender.send(websocket_message).await,
let send_timeout = self.send_timeout;
let send = async {
return match &mut self.inner {
WebSocketSinkKind::Client(sender) => sender.send(websocket_message).await,
WebSocketSinkKind::Server(sender) => sender.send(websocket_message).await,
};
};
let result = tokio::time::timeout(send_timeout, send).await;
return match result {
Ok(()) => {
Ok(Ok(())) => {
tracing::trace!(target: TRACING_TARGET, payload_len = payload_len, "binary WebSocket payload sent");
Ok(())
},
Err(error) => {
Ok(Err(error)) => {
let mapped = map_stream_error(error);
tracing::warn!(target: TRACING_TARGET, kind = %mapped.kind(), detail = mapped.detail(), "WebSocket send failed");
Err(mapped)
},
Err(_) => {
self.send_timed_out = true;
let error = timeout_error("WebSocket send", send_timeout);
tracing::warn!(target: TRACING_TARGET, timeout_ms = duration_millis(send_timeout), "WebSocket send timed out");
Err(error)
},
};
});
}
fn close(&mut self) -> Self::CloseFuture<'_> {
return Box::pin(async move {
let result = match &mut self.inner {
WebSocketSinkKind::Client(sender) => sender.close().await,
WebSocketSinkKind::Server(sender) => sender.close().await,
let close_timeout = self.close_timeout;
let close = async {
return match &mut self.inner {
WebSocketSinkKind::Client(sender) => sender.close().await,
WebSocketSinkKind::Server(sender) => sender.close().await,
};
};
let result = tokio::time::timeout(close_timeout, close).await;
return match result {
Ok(()) => {
Ok(Ok(())) => {
tracing::debug!(target: TRACING_TARGET, "local WebSocket close initiated");
Ok(())
},
Err(error) => {
Ok(Err(error)) => {
let mapped = map_stream_error(error);
tracing::warn!(target: TRACING_TARGET, kind = %mapped.kind(), detail = mapped.detail(), "WebSocket close failed");
Err(mapped)
},
Err(_) => {
let error = timeout_error("WebSocket close", close_timeout);
tracing::warn!(target: TRACING_TARGET, timeout_ms = duration_millis(close_timeout), "WebSocket close timed out");
Err(error)
},
};
});
}
@@ -188,11 +251,20 @@ impl game_realtime_transport_lib::RealtimeReceiver for WebSocketReceiver {
pub struct WebSocketListener {
listener: tokio::net::TcpListener,
local_addr: std::net::SocketAddr,
config: crate::WebSocketConfig,
}
impl WebSocketListener {
/// Binds a WebSocket listener to one concrete socket address.
/// Binds a WebSocket listener with the baseline configuration.
pub async fn bind(address: std::net::SocketAddr) -> Result<Self, game_realtime_transport_lib::TransportError> {
return Self::bind_with_config(address, crate::WebSocketConfig::default()).await;
}
/// Binds a WebSocket listener with explicit product-facing limits and deadlines.
pub async fn bind_with_config(address: std::net::SocketAddr, config: crate::WebSocketConfig) -> Result<Self, game_realtime_transport_lib::TransportError> {
if let Err(error) = config.validate() {
return Err(error);
}
let listener = match tokio::net::TcpListener::bind(address).await {
Ok(value) => value,
Err(error) => {
@@ -210,7 +282,7 @@ impl WebSocketListener {
},
};
tracing::info!(target: TRACING_TARGET, address = %local_addr, "WebSocket listener bound");
return Ok(Self { listener, local_addr });
return Ok(Self { listener, local_addr, config });
}
/// Returns the concrete local socket address, including an ephemeral port selected by the OS.
@@ -219,7 +291,7 @@ impl WebSocketListener {
return self.local_addr;
}
/// Accepts one TCP peer and completes the server-side WebSocket handshake.
/// Accepts one TCP peer and completes a bounded server-side WebSocket handshake.
pub async fn accept(&self) -> Result<crate::WebSocketConnection, game_realtime_transport_lib::TransportError> {
let (stream, peer_addr) = match self.listener.accept().await {
Ok(value) => value,
@@ -229,31 +301,51 @@ impl WebSocketListener {
return Err(mapped);
},
};
let websocket = match tokio_tungstenite::accept_async(stream).await {
Ok(value) => value,
Err(error) => {
let handshake = tokio_tungstenite::accept_async_with_config(stream, Some(tungstenite_config(&self.config)));
let result = tokio::time::timeout(self.config.connect_timeout(), handshake).await;
let websocket = match result {
Ok(Ok(value)) => value,
Ok(Err(error)) => {
let mapped = game_realtime_transport_lib::TransportError::new(game_realtime_transport_lib::TransportErrorKind::Accept, error.to_string());
tracing::warn!(target: TRACING_TARGET, peer = %peer_addr, detail = mapped.detail(), "WebSocket server handshake failed");
return Err(mapped);
},
Err(_) => {
let error = timeout_error("WebSocket server handshake", self.config.connect_timeout());
tracing::warn!(target: TRACING_TARGET, peer = %peer_addr, timeout_ms = duration_millis(self.config.connect_timeout()), "WebSocket server handshake timed out");
return Err(error);
},
};
tracing::info!(target: TRACING_TARGET, peer = %peer_addr, "WebSocket peer accepted");
return Ok(crate::WebSocketConnection::from_server(websocket));
return Ok(crate::WebSocketConnection::from_server(websocket, self.config));
}
}
/// Connects a client to one plain `ws://` endpoint and completes the WebSocket handshake.
/// Connects a client to one plain `ws://` endpoint with the baseline configuration.
pub async fn connect(endpoint: &str) -> Result<crate::WebSocketConnection, game_realtime_transport_lib::TransportError> {
return crate::connect_with_config(endpoint, crate::WebSocketConfig::default()).await;
}
/// Connects a client to one plain `ws://` endpoint with explicit limits and deadlines.
pub async fn connect_with_config(
endpoint: &str,
config: crate::WebSocketConfig,
) -> Result<crate::WebSocketConnection, game_realtime_transport_lib::TransportError> {
if !endpoint.starts_with("ws://") {
return Err(game_realtime_transport_lib::TransportError::new(
game_realtime_transport_lib::TransportErrorKind::InvalidConfiguration,
"the baseline WebSocket backend accepts only ws:// endpoints",
));
}
if let Err(error) = config.validate() {
return Err(error);
}
tracing::debug!(target: TRACING_TARGET, endpoint = endpoint, "connecting WebSocket client");
let (stream, _) = match tokio_tungstenite::connect_async(endpoint).await {
Ok(value) => value,
Err(error) => {
let handshake = tokio_tungstenite::connect_async_with_config(endpoint, Some(tungstenite_config(&config)), false);
let result = tokio::time::timeout(config.connect_timeout(), handshake).await;
let (stream, _) = match result {
Ok(Ok(value)) => value,
Ok(Err(error)) => {
let mapped = map_connect_error(error);
tracing::warn!(
target: TRACING_TARGET,
@@ -264,9 +356,33 @@ pub async fn connect(endpoint: &str) -> Result<crate::WebSocketConnection, game_
);
return Err(mapped);
},
Err(_) => {
let error = timeout_error("WebSocket client connect", config.connect_timeout());
tracing::warn!(target: TRACING_TARGET, endpoint = endpoint, timeout_ms = duration_millis(config.connect_timeout()), "WebSocket client connect timed out");
return Err(error);
},
};
tracing::info!(target: TRACING_TARGET, endpoint = endpoint, "WebSocket client connected");
return Ok(crate::WebSocketConnection::from_client(stream));
return Ok(crate::WebSocketConnection::from_client(stream, config));
}
fn tungstenite_config(config: &crate::WebSocketConfig) -> tokio_tungstenite::tungstenite::protocol::WebSocketConfig {
return tokio_tungstenite::tungstenite::protocol::WebSocketConfig::default()
.write_buffer_size(config.write_buffer_size())
.max_write_buffer_size(config.max_write_buffer_size())
.max_message_size(Some(config.max_message_size()))
.max_frame_size(Some(config.max_frame_size()));
}
fn timeout_error(operation: &str, timeout: std::time::Duration) -> game_realtime_transport_lib::TransportError {
return game_realtime_transport_lib::TransportError::new(
game_realtime_transport_lib::TransportErrorKind::Timeout,
format!("{operation} exceeded configured deadline of {} ms", duration_millis(timeout)),
);
}
fn duration_millis(duration: std::time::Duration) -> u128 {
return duration.as_millis();
}
fn map_connect_error(error: tokio_tungstenite::tungstenite::Error) -> game_realtime_transport_lib::TransportError {
@@ -289,3 +405,7 @@ fn map_stream_error(error: tokio_tungstenite::tungstenite::Error) -> game_realti
};
return game_realtime_transport_lib::TransportError::new(kind, error.to_string());
}
#[cfg(test)]
#[path = "../unit_tests/websocket.rs"]
mod tests;