From 711ed432bb0a9f3676aea8fea07d2acacf905742 Mon Sep 17 00:00:00 2001 From: Leynos Date: Wed, 6 Aug 2025 01:13:23 +0100 Subject: [PATCH 1/6] Extract config tests into dedicated module --- src/server/config.rs | 425 ------------------------------------- src/server/config/mod.rs | 225 ++++++++++++++++++++ src/server/config/tests.rs | 205 ++++++++++++++++++ 3 files changed, 430 insertions(+), 425 deletions(-) delete mode 100644 src/server/config.rs create mode 100644 src/server/config/mod.rs create mode 100644 src/server/config/tests.rs diff --git a/src/server/config.rs b/src/server/config.rs deleted file mode 100644 index e0c08a09..00000000 --- a/src/server/config.rs +++ /dev/null @@ -1,425 +0,0 @@ -//! Configuration utilities for [`WireframeServer`]. - -use core::marker::PhantomData; -use std::{ - io, - net::{SocketAddr, TcpListener as StdTcpListener}, - sync::Arc, -}; - -use bincode::error::DecodeError; -use futures::future::BoxFuture; -use tokio::{net::TcpListener, sync::oneshot}; - -use super::{Bound, PreambleCallback, PreambleErrorCallback, Unbound, WireframeServer}; -use crate::{app::WireframeApp, preamble::Preamble}; - -impl WireframeServer -where - F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, -{ - /// Create a new `WireframeServer` from the given application factory. - /// - /// The worker count defaults to the number of available CPU cores (or 1 if this cannot be - /// determined). The TCP listener is unset; call [`bind`](Self::bind) before running the - /// server. - #[must_use] - pub fn new(factory: F) -> Self { - let workers = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get); - Self { - factory, - workers, - on_preamble_success: None, - on_preamble_failure: None, - ready_tx: None, - state: Unbound, - _preamble: PhantomData, - } - } -} - -impl WireframeServer -where - F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, -{ - /// Converts the server to use a custom preamble type for incoming connections. - /// - /// Calling this method drops any previously configured preamble decode callbacks. - #[must_use] - pub fn with_preamble

(self) -> WireframeServer - where - P: Preamble, - { - WireframeServer { - factory: self.factory, - workers: self.workers, - on_preamble_success: None, - on_preamble_failure: None, - ready_tx: None, - state: self.state, - _preamble: PhantomData, - } - } -} - -impl WireframeServer -where - F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, - T: Preamble, -{ - /// Set the number of worker tasks to spawn for the server. - #[must_use] - pub fn workers(mut self, count: usize) -> Self { - self.workers = count.max(1); - self - } - - /// Register a callback invoked when the connection preamble decodes successfully. - #[must_use] - pub fn on_preamble_decode_success(mut self, handler: H) -> Self - where - H: for<'a> Fn(&'a T, &'a mut tokio::net::TcpStream) -> BoxFuture<'a, io::Result<()>> - + Send - + Sync - + 'static, - { - self.on_preamble_success = Some(Arc::new(handler)); - self - } - - /// Register a callback invoked when the connection preamble fails to decode. - #[must_use] - pub fn on_preamble_decode_failure(mut self, handler: H) -> Self - where - H: Fn(&DecodeError) + Send + Sync + 'static, - { - self.on_preamble_failure = Some(Arc::new(handler)); - self - } - - /// Configure a channel used to signal when the server is ready to accept connections. - #[must_use] - pub fn ready_signal(mut self, tx: oneshot::Sender<()>) -> Self { - self.ready_tx = Some(tx); - self - } - - /// Returns the configured number of worker tasks for the server. - #[inline] - #[must_use] - pub const fn worker_count(&self) -> usize { self.workers } - - /// Delegate binding to [`bind_std_listener`] after extracting fields. - /// - /// The public `bind` and `bind_listener` methods merely prepare the - /// [`StdTcpListener`] before calling this helper. - fn bind_with_std_listener( - self, - std_listener: StdTcpListener, - ) -> io::Result> { - let Self { - factory, - workers, - on_preamble_success, - on_preamble_failure, - ready_tx, - .. - } = self; - bind_std_listener( - factory, - workers, - on_preamble_success, - on_preamble_failure, - ready_tx, - std_listener, - ) - } -} - -impl WireframeServer -where - F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, - T: Preamble, -{ - /// Get the socket address the server is bound to, if available. - #[must_use] - pub const fn local_addr(&self) -> Option { None } - - /// Bind the server to the given address and create a listener. - /// - /// # Errors - /// Returns an `io::Error` if binding or configuring the listener fails. - pub fn bind(self, addr: SocketAddr) -> io::Result> { - let std_listener = StdTcpListener::bind(addr)?; - self.bind_with_std_listener(std_listener) - } - - /// Bind the server to an existing standard TCP listener. - /// - /// # Errors - /// Returns an [`io::Error`] if configuring the listener fails. - pub fn bind_listener( - self, - listener: StdTcpListener, - ) -> io::Result> { - self.bind_with_std_listener(listener) - } -} - -impl WireframeServer -where - F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, - T: Preamble, -{ - /// Get the socket address the server is bound to. - #[must_use] - pub fn local_addr(&self) -> Option { self.state.listener.local_addr().ok() } - - /// Rebind the server to a new address. - /// - /// # Errors - /// Returns an `io::Error` if binding or configuring the listener fails. - pub fn bind(self, addr: SocketAddr) -> io::Result { - let std_listener = StdTcpListener::bind(addr)?; - self.bind_with_std_listener(std_listener) - } - - /// Rebind the server to an existing standard TCP listener. - /// - /// # Errors - /// Returns an [`io::Error`] if configuring the listener fails. - pub fn bind_listener(self, listener: StdTcpListener) -> io::Result { - self.bind_with_std_listener(listener) - } -} - -#[allow(clippy::too_many_arguments)] -fn bind_std_listener( - factory: F, - workers: usize, - on_preamble_success: Option>, - on_preamble_failure: Option, - ready_tx: Option>, - std_listener: StdTcpListener, -) -> io::Result> -where - F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, - T: Preamble, -{ - std_listener.set_nonblocking(true)?; - let listener = TcpListener::from_std(std_listener)?; - Ok(WireframeServer { - factory, - workers, - on_preamble_success, - on_preamble_failure, - ready_tx, - state: Bound { - listener: Arc::new(listener), - }, - _preamble: PhantomData, - }) -} - -#[cfg(test)] -mod tests { - use std::{ - net::SocketAddr, - sync::{ - Arc, - atomic::{AtomicUsize, Ordering}, - }, - }; - - use rstest::rstest; - - use super::*; - use crate::server::test_util::{ - TestPreamble, - bind_server, - factory, - free_port, - server_with_preamble, - }; - - fn expected_default_worker_count() -> usize { - // Mirror the default worker logic to keep tests aligned with `WireframeServer::new`. - std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get) - } - - #[rstest] - fn test_new_server_creation( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - ) { - let server = WireframeServer::new(factory); - assert!(server.worker_count() >= 1 && server.local_addr().is_none()); - } - - #[rstest] - fn test_new_server_default_worker_count( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - ) { - let server = WireframeServer::new(factory); - assert_eq!(server.worker_count(), expected_default_worker_count()); - } - - #[rstest] - fn test_workers_configuration( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - ) { - let mut server = WireframeServer::new(factory); - server = server.workers(4); - assert_eq!(server.worker_count(), 4); - server = server.workers(100); - assert_eq!(server.worker_count(), 100); - assert_eq!(server.workers(0).worker_count(), 1); - } - - #[rstest] - fn test_with_preamble_type_conversion( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - ) { - let server = WireframeServer::new(factory).with_preamble::(); - assert_eq!(server.worker_count(), expected_default_worker_count()); - } - - #[rstest] - #[tokio::test] - async fn test_bind_success( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - free_port: SocketAddr, - ) { - let local_addr = WireframeServer::new(factory) - .bind(free_port) - .expect("Failed to bind") - .local_addr() - .expect("local address missing"); - assert_eq!(local_addr.ip(), free_port.ip()); - } - - #[rstest] - fn test_local_addr_before_bind( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - ) { - assert!(WireframeServer::new(factory).local_addr().is_none()); - } - - #[rstest] - #[tokio::test] - async fn test_local_addr_after_bind( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - free_port: SocketAddr, - ) { - let local_addr = bind_server(factory, free_port).local_addr().unwrap(); - assert_eq!(local_addr.ip(), free_port.ip()); - } - - #[rstest] - #[case("success")] - #[case("failure")] - #[tokio::test] - async fn test_preamble_callback_registration( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - #[case] callback_type: &str, - ) { - let counter = Arc::new(AtomicUsize::new(0)); - let c = counter.clone(); - - let server = server_with_preamble(factory); - let server = match callback_type { - "success" => server.on_preamble_decode_success(move |_p: &TestPreamble, _| { - let c = c.clone(); - Box::pin(async move { - c.fetch_add(1, Ordering::SeqCst); - Ok(()) - }) - }), - "failure" => server.on_preamble_decode_failure(move |_err: &DecodeError| { - c.fetch_add(1, Ordering::SeqCst); - }), - _ => panic!("Invalid callback type"), - }; - - assert_eq!(counter.load(Ordering::SeqCst), 0); - match callback_type { - "success" => assert!(server.on_preamble_success.is_some()), - "failure" => assert!(server.on_preamble_failure.is_some()), - _ => unreachable!(), - } - } - - #[rstest] - #[tokio::test] - async fn test_method_chaining( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - free_port: SocketAddr, - ) { - let callback_invoked = Arc::new(AtomicUsize::new(0)); - let counter = callback_invoked.clone(); - let server = WireframeServer::new(factory) - .workers(2) - .with_preamble::() - .on_preamble_decode_success(move |_p: &TestPreamble, _| { - let c = counter.clone(); - Box::pin(async move { - c.fetch_add(1, Ordering::SeqCst); - Ok(()) - }) - }) - .on_preamble_decode_failure(|_: &DecodeError| {}) - .bind(free_port) - .expect("Failed to bind"); - assert_eq!(server.worker_count(), 2); - assert!(server.local_addr().is_some()); - assert_eq!(callback_invoked.load(Ordering::SeqCst), 0); - } - - #[rstest] - #[tokio::test] - async fn test_server_configuration_persistence( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - free_port: SocketAddr, - ) { - let server = WireframeServer::new(factory) - .workers(5) - .bind(free_port) - .expect("Failed to bind"); - assert_eq!(server.worker_count(), 5); - assert!(server.local_addr().is_some()); - } - - #[rstest] - fn test_extreme_worker_counts( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - ) { - let mut server = WireframeServer::new(factory); - server = server.workers(usize::MAX); - assert_eq!(server.worker_count(), usize::MAX); - assert_eq!(server.workers(0).worker_count(), 1); - } - - #[rstest] - #[tokio::test] - async fn test_bind_to_multiple_addresses( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - free_port: SocketAddr, - ) { - let listener2 = - std::net::TcpListener::bind(SocketAddr::new(std::net::Ipv4Addr::LOCALHOST.into(), 0)) - .expect("failed to bind second listener"); - let addr2 = listener2 - .local_addr() - .expect("failed to get second listener address"); - drop(listener2); - - let server = WireframeServer::new(factory); - let server = server - .bind(free_port) - .expect("Failed to bind first address"); - let first = server.local_addr().expect("first bound address missing"); - let server = server.bind(addr2).expect("Failed to bind second address"); - let second = server.local_addr().expect("second bound address missing"); - assert_ne!(first.port(), second.port()); - assert_eq!(second.ip(), addr2.ip()); - } -} diff --git a/src/server/config/mod.rs b/src/server/config/mod.rs new file mode 100644 index 00000000..a4519cd4 --- /dev/null +++ b/src/server/config/mod.rs @@ -0,0 +1,225 @@ +//! Configuration utilities for [`WireframeServer`]. + +use core::marker::PhantomData; +use std::{ + io, + net::{SocketAddr, TcpListener as StdTcpListener}, + sync::Arc, +}; + +use bincode::error::DecodeError; +use futures::future::BoxFuture; +use tokio::{net::TcpListener, sync::oneshot}; + +use super::{Bound, PreambleCallback, PreambleErrorCallback, Unbound, WireframeServer}; +use crate::{app::WireframeApp, preamble::Preamble}; + +#[cfg(test)] +mod tests; + +impl WireframeServer +where + F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, +{ + /// Create a new `WireframeServer` from the given application factory. + /// + /// The worker count defaults to the number of available CPU cores (or 1 if this cannot be + /// determined). The TCP listener is unset; call [`bind`](Self::bind) before running the + /// server. + #[must_use] + pub fn new(factory: F) -> Self { + let workers = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get); + Self { + factory, + workers, + on_preamble_success: None, + on_preamble_failure: None, + ready_tx: None, + state: Unbound, + _preamble: PhantomData, + } + } +} + +impl WireframeServer +where + F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, +{ + /// Converts the server to use a custom preamble type for incoming connections. + /// + /// Calling this method drops any previously configured preamble decode callbacks. + #[must_use] + pub fn with_preamble

(self) -> WireframeServer + where + P: Preamble, + { + WireframeServer { + factory: self.factory, + workers: self.workers, + on_preamble_success: None, + on_preamble_failure: None, + ready_tx: None, + state: self.state, + _preamble: PhantomData, + } + } +} + +impl WireframeServer +where + F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, + T: Preamble, +{ + /// Set the number of worker tasks to spawn for the server. + #[must_use] + pub fn workers(mut self, count: usize) -> Self { + self.workers = count.max(1); + self + } + + /// Register a callback invoked when the connection preamble decodes successfully. + #[must_use] + pub fn on_preamble_decode_success(mut self, handler: H) -> Self + where + H: for<'a> Fn(&'a T, &'a mut tokio::net::TcpStream) -> BoxFuture<'a, io::Result<()>> + + Send + + Sync + + 'static, + { + self.on_preamble_success = Some(Arc::new(handler)); + self + } + + /// Register a callback invoked when the connection preamble fails to decode. + #[must_use] + pub fn on_preamble_decode_failure(mut self, handler: H) -> Self + where + H: Fn(&DecodeError) + Send + Sync + 'static, + { + self.on_preamble_failure = Some(Arc::new(handler)); + self + } + + /// Configure a channel used to signal when the server is ready to accept connections. + #[must_use] + pub fn ready_signal(mut self, tx: oneshot::Sender<()>) -> Self { + self.ready_tx = Some(tx); + self + } + + /// Returns the configured number of worker tasks for the server. + #[inline] + #[must_use] + pub const fn worker_count(&self) -> usize { self.workers } + + /// Delegate binding to [`bind_std_listener`] after extracting fields. + /// + /// The public `bind` and `bind_listener` methods merely prepare the + /// [`StdTcpListener`] before calling this helper. + fn bind_with_std_listener( + self, + std_listener: StdTcpListener, + ) -> io::Result> { + let Self { + factory, + workers, + on_preamble_success, + on_preamble_failure, + ready_tx, + .. + } = self; + bind_std_listener( + factory, + workers, + on_preamble_success, + on_preamble_failure, + ready_tx, + std_listener, + ) + } +} + +impl WireframeServer +where + F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, + T: Preamble, +{ + /// Get the socket address the server is bound to, if available. + #[must_use] + pub const fn local_addr(&self) -> Option { None } + + /// Bind the server to the given address and create a listener. + /// + /// # Errors + /// Returns an `io::Error` if binding or configuring the listener fails. + pub fn bind(self, addr: SocketAddr) -> io::Result> { + let std_listener = StdTcpListener::bind(addr)?; + self.bind_with_std_listener(std_listener) + } + + /// Bind the server to an existing standard TCP listener. + /// + /// # Errors + /// Returns an [`io::Error`] if configuring the listener fails. + pub fn bind_listener( + self, + listener: StdTcpListener, + ) -> io::Result> { + self.bind_with_std_listener(listener) + } +} + +impl WireframeServer +where + F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, + T: Preamble, +{ + /// Get the socket address the server is bound to. + #[must_use] + pub fn local_addr(&self) -> Option { self.state.listener.local_addr().ok() } + + /// Rebind the server to a new address. + /// + /// # Errors + /// Returns an `io::Error` if binding or configuring the listener fails. + pub fn bind(self, addr: SocketAddr) -> io::Result { + let std_listener = StdTcpListener::bind(addr)?; + self.bind_with_std_listener(std_listener) + } + + /// Rebind the server to an existing standard TCP listener. + /// + /// # Errors + /// Returns an [`io::Error`] if configuring the listener fails. + pub fn bind_listener(self, listener: StdTcpListener) -> io::Result { + self.bind_with_std_listener(listener) + } +} + +#[allow(clippy::too_many_arguments)] +fn bind_std_listener( + factory: F, + workers: usize, + on_preamble_success: Option>, + on_preamble_failure: Option, + ready_tx: Option>, + std_listener: StdTcpListener, +) -> io::Result> +where + F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, + T: Preamble, +{ + std_listener.set_nonblocking(true)?; + let listener = TcpListener::from_std(std_listener)?; + Ok(WireframeServer { + factory, + workers, + on_preamble_success, + on_preamble_failure, + ready_tx, + state: Bound { + listener: Arc::new(listener), + }, + _preamble: PhantomData, + }) +} diff --git a/src/server/config/tests.rs b/src/server/config/tests.rs new file mode 100644 index 00000000..5297b3eb --- /dev/null +++ b/src/server/config/tests.rs @@ -0,0 +1,205 @@ +//! Tests for server configuration utilities. + +use std::{ + net::SocketAddr, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, +}; + +use rstest::rstest; + +use super::*; +use crate::server::test_util::{ + TestPreamble, + bind_server, + factory, + free_port, + server_with_preamble, +}; + +fn expected_default_worker_count() -> usize { + // Mirror the default worker logic to keep tests aligned with `WireframeServer::new`. + std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get) +} + +#[rstest] +fn test_new_server_creation(factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static) { + let server = WireframeServer::new(factory); + assert!(server.worker_count() >= 1 && server.local_addr().is_none()); +} + +#[rstest] +fn test_new_server_default_worker_count( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, +) { + let server = WireframeServer::new(factory); + assert_eq!(server.worker_count(), expected_default_worker_count()); +} + +#[rstest] +fn test_workers_configuration(factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static) { + let mut server = WireframeServer::new(factory); + server = server.workers(4); + assert_eq!(server.worker_count(), 4); + server = server.workers(100); + assert_eq!(server.worker_count(), 100); + assert_eq!(server.workers(0).worker_count(), 1); +} + +#[rstest] +fn test_with_preamble_type_conversion( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, +) { + let server = WireframeServer::new(factory).with_preamble::(); + assert_eq!(server.worker_count(), expected_default_worker_count()); +} + +#[rstest] +#[tokio::test] +async fn test_bind_success( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, + free_port: SocketAddr, +) { + let local_addr = WireframeServer::new(factory) + .bind(free_port) + .expect("Failed to bind") + .local_addr() + .expect("local address missing"); + assert_eq!(local_addr.ip(), free_port.ip()); +} + +#[rstest] +fn test_local_addr_before_bind(factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static) { + assert!(WireframeServer::new(factory).local_addr().is_none()); +} + +#[rstest] +#[tokio::test] +async fn test_local_addr_after_bind( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, + free_port: SocketAddr, +) { + let local_addr = bind_server(factory, free_port).local_addr().unwrap(); + assert_eq!(local_addr.ip(), free_port.ip()); +} + +#[rstest] +#[tokio::test] +async fn test_preamble_success_callback( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, +) { + let counter = Arc::new(AtomicUsize::new(0)); + let c = counter.clone(); + let server = + server_with_preamble(factory).on_preamble_decode_success(move |_p: &TestPreamble, _| { + let c = c.clone(); + Box::pin(async move { + c.fetch_add(1, Ordering::SeqCst); + Ok(()) + }) + }); + assert_eq!(counter.load(Ordering::SeqCst), 0); + assert!(server.on_preamble_success.is_some()); +} + +#[rstest] +#[tokio::test] +async fn test_preamble_failure_callback( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, +) { + let counter = Arc::new(AtomicUsize::new(0)); + let c = counter.clone(); + let server = + server_with_preamble(factory).on_preamble_decode_failure(move |_err: &DecodeError| { + c.fetch_add(1, Ordering::SeqCst); + }); + assert_eq!(counter.load(Ordering::SeqCst), 0); + assert!(server.on_preamble_failure.is_some()); +} + +#[rstest] +#[tokio::test] +async fn test_method_chaining( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, + free_port: SocketAddr, +) { + let counter = Arc::new(AtomicUsize::new(0)); + let c = counter.clone(); + let server = WireframeServer::new(factory) + .workers(2) + .with_preamble::() + .on_preamble_decode_success(move |_p: &TestPreamble, _| { + let c = c.clone(); + Box::pin(async move { + c.fetch_add(1, Ordering::SeqCst); + Ok(()) + }) + }) + .on_preamble_decode_failure(|_: &DecodeError| {}) + .bind(free_port) + .expect("Failed to bind"); + assert_eq!(server.worker_count(), 2); + assert!(server.local_addr().is_some()); + assert!(server.on_preamble_success.is_some() && server.on_preamble_failure.is_some()); +} + +#[rstest] +#[tokio::test] +async fn test_server_configuration_persistence( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, + free_port: SocketAddr, +) { + let server = WireframeServer::new(factory) + .workers(5) + .bind(free_port) + .expect("Failed to bind"); + assert_eq!(server.worker_count(), 5); + assert!(server.local_addr().is_some()); +} + +#[rstest] +fn test_preamble_callbacks_reset_on_type_change( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, +) { + let server = WireframeServer::new(factory) + .on_preamble_decode_success(|&(), _| Box::pin(async { Ok(()) })) + .on_preamble_decode_failure(|_: &DecodeError| {}); + assert!(server.on_preamble_success.is_some() && server.on_preamble_failure.is_some()); + let server = server.with_preamble::(); + assert!(server.on_preamble_success.is_none() && server.on_preamble_failure.is_none()); +} + +#[rstest] +fn test_extreme_worker_counts(factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static) { + let mut server = WireframeServer::new(factory); + server = server.workers(usize::MAX); + assert_eq!(server.worker_count(), usize::MAX); + assert_eq!(server.workers(0).worker_count(), 1); +} + +#[rstest] +#[tokio::test] +async fn test_bind_to_multiple_addresses( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, + free_port: SocketAddr, +) { + let addr2 = { + let listener = + std::net::TcpListener::bind(SocketAddr::new(std::net::Ipv4Addr::LOCALHOST.into(), 0)) + .expect("failed to bind second listener"); + listener + .local_addr() + .expect("failed to get second listener address") + }; + let server = WireframeServer::new(factory); + let server = server + .bind(free_port) + .expect("Failed to bind first address"); + let first = server.local_addr().expect("first bound address missing"); + let server = server.bind(addr2).expect("Failed to bind second address"); + let second = server.local_addr().expect("second bound address missing"); + assert_ne!(first.port(), second.port()); + assert_eq!(second.ip(), addr2.ip()); +} From 5b9f5eea519f21abb32c27e8698217381f6fb93a Mon Sep 17 00:00:00 2001 From: Leynos Date: Wed, 6 Aug 2025 07:58:22 +0100 Subject: [PATCH 2/6] Parametrize preamble callback tests --- src/server/config/mod.rs | 6 +++- src/server/config/tests.rs | 63 +++++++++++++++++++------------------- 2 files changed, 37 insertions(+), 32 deletions(-) diff --git a/src/server/config/mod.rs b/src/server/config/mod.rs index a4519cd4..bb682875 100644 --- a/src/server/config/mod.rs +++ b/src/server/config/mod.rs @@ -1,4 +1,9 @@ //! Configuration utilities for [`WireframeServer`]. +//! +//! Provides a fluent builder for configuring `WireframeServer` instances. +//! The builder exposes worker count tuning, preamble callbacks, and TCP +//! binding. Phantom types capture the bind state at compile time so invalid +//! transitions, such as rebinding an uninitialised server, fail to compile. use core::marker::PhantomData; use std::{ @@ -196,7 +201,6 @@ where } } -#[allow(clippy::too_many_arguments)] fn bind_std_listener( factory: F, workers: usize, diff --git a/src/server/config/tests.rs b/src/server/config/tests.rs index 5297b3eb..f479244c 100644 --- a/src/server/config/tests.rs +++ b/src/server/config/tests.rs @@ -86,37 +86,30 @@ async fn test_local_addr_after_bind( } #[rstest] +#[case("success")] +#[case("failure")] #[tokio::test] -async fn test_preamble_success_callback( +async fn test_preamble_callback_registration( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, + #[case] callback_type: &str, ) { let counter = Arc::new(AtomicUsize::new(0)); let c = counter.clone(); - let server = - server_with_preamble(factory).on_preamble_decode_success(move |_p: &TestPreamble, _| { + let server = server_with_preamble(factory); + let _server = match callback_type { + "success" => server.on_preamble_decode_success(move |_p: &TestPreamble, _| { let c = c.clone(); Box::pin(async move { c.fetch_add(1, Ordering::SeqCst); Ok(()) }) - }); - assert_eq!(counter.load(Ordering::SeqCst), 0); - assert!(server.on_preamble_success.is_some()); -} - -#[rstest] -#[tokio::test] -async fn test_preamble_failure_callback( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, -) { - let counter = Arc::new(AtomicUsize::new(0)); - let c = counter.clone(); - let server = - server_with_preamble(factory).on_preamble_decode_failure(move |_err: &DecodeError| { + }), + "failure" => server.on_preamble_decode_failure(move |_err: &DecodeError| { c.fetch_add(1, Ordering::SeqCst); - }); + }), + _ => unreachable!("invalid case"), + }; assert_eq!(counter.load(Ordering::SeqCst), 0); - assert!(server.on_preamble_failure.is_some()); } #[rstest] @@ -142,7 +135,6 @@ async fn test_method_chaining( .expect("Failed to bind"); assert_eq!(server.worker_count(), 2); assert!(server.local_addr().is_some()); - assert!(server.on_preamble_success.is_some() && server.on_preamble_failure.is_some()); } #[rstest] @@ -166,9 +158,19 @@ fn test_preamble_callbacks_reset_on_type_change( let server = WireframeServer::new(factory) .on_preamble_decode_success(|&(), _| Box::pin(async { Ok(()) })) .on_preamble_decode_failure(|_: &DecodeError| {}); - assert!(server.on_preamble_success.is_some() && server.on_preamble_failure.is_some()); - let server = server.with_preamble::(); - assert!(server.on_preamble_success.is_none() && server.on_preamble_failure.is_none()); + let counter = Arc::new(AtomicUsize::new(0)); + let c = counter.clone(); + let _server = server + .with_preamble::() + .on_preamble_decode_success(move |_p: &TestPreamble, _| { + let c = c.clone(); + Box::pin(async move { + c.fetch_add(1, Ordering::SeqCst); + Ok(()) + }) + }) + .on_preamble_decode_failure(|_: &DecodeError| {}); + assert_eq!(counter.load(Ordering::SeqCst), 0); } #[rstest] @@ -185,19 +187,18 @@ async fn test_bind_to_multiple_addresses( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, free_port: SocketAddr, ) { - let addr2 = { - let listener = - std::net::TcpListener::bind(SocketAddr::new(std::net::Ipv4Addr::LOCALHOST.into(), 0)) - .expect("failed to bind second listener"); - listener - .local_addr() - .expect("failed to get second listener address") - }; + let listener2 = + std::net::TcpListener::bind(SocketAddr::new(std::net::Ipv4Addr::LOCALHOST.into(), 0)) + .expect("failed to bind second listener"); + let addr2 = listener2 + .local_addr() + .expect("failed to get second listener address"); let server = WireframeServer::new(factory); let server = server .bind(free_port) .expect("Failed to bind first address"); let first = server.local_addr().expect("first bound address missing"); + drop(listener2); let server = server.bind(addr2).expect("Failed to bind second address"); let second = server.local_addr().expect("second bound address missing"); assert_ne!(first.port(), second.port()); From fea3ef68ce83a82efed61b25500409c69f0157ae Mon Sep 17 00:00:00 2001 From: Leynos Date: Wed, 6 Aug 2025 08:25:40 +0100 Subject: [PATCH 3/6] Clarify config tests and add usage examples --- src/server/config/mod.rs | 34 +++++++++++++++++++---- src/server/config/tests.rs | 57 ++++++++++++++++++++++++++++++-------- 2 files changed, 75 insertions(+), 16 deletions(-) diff --git a/src/server/config/mod.rs b/src/server/config/mod.rs index bb682875..4e8143a3 100644 --- a/src/server/config/mod.rs +++ b/src/server/config/mod.rs @@ -28,9 +28,18 @@ where { /// Create a new `WireframeServer` from the given application factory. /// - /// The worker count defaults to the number of available CPU cores (or 1 if this cannot be - /// determined). The TCP listener is unset; call [`bind`](Self::bind) before running the - /// server. + /// The worker count defaults to the number of available CPU cores (or 1 if + /// this cannot be determined). The TCP listener is unset; call + /// [`bind`](Self::bind) before running the server. + /// + /// # Examples + /// + /// ``` + /// use wireframe::{app::WireframeApp, server::WireframeServer}; + /// + /// let server = WireframeServer::new(|| WireframeApp::default()); + /// assert!(server.worker_count() >= 1); + /// ``` #[must_use] pub fn new(factory: F) -> Self { let workers = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get); @@ -50,9 +59,24 @@ impl WireframeServer where F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, { - /// Converts the server to use a custom preamble type for incoming connections. + /// Converts the server to use a custom preamble type for incoming + /// connections. + /// + /// Calling this method drops any previously configured preamble decode + /// callbacks. + /// + /// # Examples + /// + /// ``` + /// use bincode::{Decode, Encode}; + /// use wireframe::{app::WireframeApp, preamble::Preamble, server::WireframeServer}; + /// + /// #[derive(Encode, Decode)] + /// struct MyPreamble; + /// impl Preamble for MyPreamble {} /// - /// Calling this method drops any previously configured preamble decode callbacks. + /// let server = WireframeServer::new(|| WireframeApp::default()).with_preamble::(); + /// ``` #[must_use] pub fn with_preamble

(self) -> WireframeServer where diff --git a/src/server/config/tests.rs b/src/server/config/tests.rs index f479244c..476d44e5 100644 --- a/src/server/config/tests.rs +++ b/src/server/config/tests.rs @@ -1,7 +1,12 @@ //! Tests for server configuration utilities. +//! +//! This module exercises the `WireframeServer` builder, covering worker counts, +//! binding behaviour, preamble handling, callback registration, and method +//! chaining. Fixtures from `test_util` provide shared setup and parameterised +//! cases via `rstest`. use std::{ - net::SocketAddr, + net::{Ipv4Addr, SocketAddr}, sync::{ Arc, atomic::{AtomicUsize, Ordering}, @@ -19,15 +24,11 @@ use crate::server::test_util::{ server_with_preamble, }; -fn expected_default_worker_count() -> usize { - // Mirror the default worker logic to keep tests aligned with `WireframeServer::new`. - std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get) -} - #[rstest] fn test_new_server_creation(factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static) { let server = WireframeServer::new(factory); - assert!(server.worker_count() >= 1 && server.local_addr().is_none()); + assert!(server.worker_count() >= 1); + assert!(server.local_addr().is_none()); } #[rstest] @@ -35,7 +36,8 @@ fn test_new_server_default_worker_count( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { let server = WireframeServer::new(factory); - assert_eq!(server.worker_count(), expected_default_worker_count()); + let expected = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get); + assert_eq!(server.worker_count(), expected); } #[rstest] @@ -53,7 +55,8 @@ fn test_with_preamble_type_conversion( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { let server = WireframeServer::new(factory).with_preamble::(); - assert_eq!(server.worker_count(), expected_default_worker_count()); + let expected = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get); + assert_eq!(server.worker_count(), expected); } #[rstest] @@ -96,7 +99,7 @@ async fn test_preamble_callback_registration( let counter = Arc::new(AtomicUsize::new(0)); let c = counter.clone(); let server = server_with_preamble(factory); - let _server = match callback_type { + let server = match callback_type { "success" => server.on_preamble_decode_success(move |_p: &TestPreamble, _| { let c = c.clone(); Box::pin(async move { @@ -109,7 +112,39 @@ async fn test_preamble_callback_registration( }), _ => unreachable!("invalid case"), }; - assert_eq!(counter.load(Ordering::SeqCst), 0); + + match callback_type { + "success" => { + let cb = server + .on_preamble_success + .as_ref() + .expect("callback not registered"); + let addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0); + let listener = tokio::net::TcpListener::bind(addr) + .await + .expect("bind temporary listener"); + let addr = listener.local_addr().expect("read listener addr"); + let mut stream = tokio::net::TcpStream::connect(addr) + .await + .expect("connect stream"); + drop(listener); + let preamble = TestPreamble { + id: 0, + message: String::new(), + }; + cb(&preamble, &mut stream).await.expect("callback failed"); + } + "failure" => { + let cb = server + .on_preamble_failure + .as_ref() + .expect("callback not registered"); + cb(&DecodeError::OtherString(String::from("err"))); + } + _ => unreachable!("invalid case"), + } + + assert_eq!(counter.load(Ordering::SeqCst), 1); } #[rstest] From b29246d3e725dc03e610b35613ee1999941edace Mon Sep 17 00:00:00 2001 From: Leynos Date: Wed, 6 Aug 2025 18:26:00 +0100 Subject: [PATCH 4/6] Simplify server binding with optional listener --- src/server/config/mod.rs | 135 +++++++-------------------------------- src/server/mod.rs | 12 +--- src/server/runtime.rs | 7 +- src/server/test_util.rs | 4 +- 4 files changed, 31 insertions(+), 127 deletions(-) diff --git a/src/server/config/mod.rs b/src/server/config/mod.rs index 4e8143a3..dbbaf0aa 100644 --- a/src/server/config/mod.rs +++ b/src/server/config/mod.rs @@ -1,9 +1,9 @@ //! Configuration utilities for [`WireframeServer`]. //! //! Provides a fluent builder for configuring `WireframeServer` instances. -//! The builder exposes worker count tuning, preamble callbacks, and TCP -//! binding. Phantom types capture the bind state at compile time so invalid -//! transitions, such as rebinding an uninitialised server, fail to compile. +//! The builder exposes worker count tuning, preamble callbacks, ready-signal +//! configuration, and TCP binding. The server holds an optional listener so it +//! may be constructed unbound and later bound via [`bind`](WireframeServer::bind). use core::marker::PhantomData; use std::{ @@ -16,13 +16,13 @@ use bincode::error::DecodeError; use futures::future::BoxFuture; use tokio::{net::TcpListener, sync::oneshot}; -use super::{Bound, PreambleCallback, PreambleErrorCallback, Unbound, WireframeServer}; +use super::WireframeServer; use crate::{app::WireframeApp, preamble::Preamble}; #[cfg(test)] mod tests; -impl WireframeServer +impl WireframeServer where F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, { @@ -49,15 +49,16 @@ where on_preamble_success: None, on_preamble_failure: None, ready_tx: None, - state: Unbound, + listener: None, _preamble: PhantomData, } } } -impl WireframeServer +impl WireframeServer where F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, + T: Preamble, { /// Converts the server to use a custom preamble type for incoming /// connections. @@ -78,7 +79,7 @@ where /// let server = WireframeServer::new(|| WireframeApp::default()).with_preamble::(); /// ``` #[must_use] - pub fn with_preamble

(self) -> WireframeServer + pub fn with_preamble

(self) -> WireframeServer where P: Preamble, { @@ -88,17 +89,11 @@ where on_preamble_success: None, on_preamble_failure: None, ready_tx: None, - state: self.state, + listener: self.listener, _preamble: PhantomData, } } -} -impl WireframeServer -where - F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, - T: Preamble, -{ /// Set the number of worker tasks to spawn for the server. #[must_use] pub fn workers(mut self, count: usize) -> Self { @@ -141,113 +136,29 @@ where #[must_use] pub const fn worker_count(&self) -> usize { self.workers } - /// Delegate binding to [`bind_std_listener`] after extracting fields. - /// - /// The public `bind` and `bind_listener` methods merely prepare the - /// [`StdTcpListener`] before calling this helper. - fn bind_with_std_listener( - self, - std_listener: StdTcpListener, - ) -> io::Result> { - let Self { - factory, - workers, - on_preamble_success, - on_preamble_failure, - ready_tx, - .. - } = self; - bind_std_listener( - factory, - workers, - on_preamble_success, - on_preamble_failure, - ready_tx, - std_listener, - ) - } -} - -impl WireframeServer -where - F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, - T: Preamble, -{ - /// Get the socket address the server is bound to, if available. + /// Returns the bound address, or `None` if not yet bound. #[must_use] - pub const fn local_addr(&self) -> Option { None } - - /// Bind the server to the given address and create a listener. - /// - /// # Errors - /// Returns an `io::Error` if binding or configuring the listener fails. - pub fn bind(self, addr: SocketAddr) -> io::Result> { - let std_listener = StdTcpListener::bind(addr)?; - self.bind_with_std_listener(std_listener) + pub fn local_addr(&self) -> Option { + self.listener.as_ref().and_then(|l| l.local_addr().ok()) } - /// Bind the server to an existing standard TCP listener. - /// - /// # Errors - /// Returns an [`io::Error`] if configuring the listener fails. - pub fn bind_listener( - self, - listener: StdTcpListener, - ) -> io::Result> { - self.bind_with_std_listener(listener) - } -} - -impl WireframeServer -where - F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, - T: Preamble, -{ - /// Get the socket address the server is bound to. - #[must_use] - pub fn local_addr(&self) -> Option { self.state.listener.local_addr().ok() } - - /// Rebind the server to a new address. + /// Bind to a fresh address. /// /// # Errors /// Returns an `io::Error` if binding or configuring the listener fails. pub fn bind(self, addr: SocketAddr) -> io::Result { - let std_listener = StdTcpListener::bind(addr)?; - self.bind_with_std_listener(std_listener) + let std = StdTcpListener::bind(addr)?; + self.bind_listener(std) } - /// Rebind the server to an existing standard TCP listener. + /// Bind to an existing `StdTcpListener`. /// /// # Errors - /// Returns an [`io::Error`] if configuring the listener fails. - pub fn bind_listener(self, listener: StdTcpListener) -> io::Result { - self.bind_with_std_listener(listener) + /// Returns an `io::Error` if configuring the listener fails. + pub fn bind_listener(mut self, std: StdTcpListener) -> io::Result { + std.set_nonblocking(true)?; + let tokio = TcpListener::from_std(std)?; + self.listener = Some(Arc::new(tokio)); + Ok(self) } } - -fn bind_std_listener( - factory: F, - workers: usize, - on_preamble_success: Option>, - on_preamble_failure: Option, - ready_tx: Option>, - std_listener: StdTcpListener, -) -> io::Result> -where - F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, - T: Preamble, -{ - std_listener.set_nonblocking(true)?; - let listener = TcpListener::from_std(std_listener)?; - Ok(WireframeServer { - factory, - workers, - on_preamble_success, - on_preamble_failure, - ready_tx, - state: Bound { - listener: Arc::new(listener), - }, - _preamble: PhantomData, - }) -} diff --git a/src/server/mod.rs b/src/server/mod.rs index 9dd08238..57b9c4ab 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -33,15 +33,7 @@ pub type PreambleErrorCallback = Arc; /// closure. The server listens for a shutdown signal using /// `tokio::signal::ctrl_c` and notifies all workers to stop /// accepting new connections. -#[doc(hidden)] -pub struct Unbound; - -#[doc(hidden)] -pub struct Bound { - pub(crate) listener: Arc, -} - -pub struct WireframeServer +pub struct WireframeServer where F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, // `Preamble` covers types implementing `BorrowDecode` for any lifetime, @@ -65,7 +57,7 @@ where /// Because only one notification may be sent, a new `ready_tx` must be /// provided each time the server is started. pub(crate) ready_tx: Option>, - pub(crate) state: S, + pub(crate) listener: Option>, pub(crate) _preamble: PhantomData, } diff --git a/src/server/runtime.rs b/src/server/runtime.rs index 6f9354a5..c9d81f85 100644 --- a/src/server/runtime.rs +++ b/src/server/runtime.rs @@ -12,7 +12,6 @@ use tokio::{ use tokio_util::{sync::CancellationToken, task::TaskTracker}; use super::{ - Bound, PreambleCallback, PreambleErrorCallback, WireframeServer, @@ -23,7 +22,7 @@ use crate::{app::WireframeApp, preamble::Preamble}; const ACCEPT_RETRY_INITIAL_DELAY: Duration = Duration::from_millis(10); const ACCEPT_RETRY_MAX_DELAY: Duration = Duration::from_secs(1); -impl WireframeServer +impl WireframeServer where F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, T: Preamble, @@ -57,10 +56,12 @@ where on_preamble_success, on_preamble_failure, ready_tx, - state: Bound { listener }, + listener, .. } = self; + let listener = listener.ok_or_else(|| io::Error::other("listener not bound"))?; + if let Some(tx) = ready_tx && tx.send(()).is_err() { diff --git a/src/server/test_util.rs b/src/server/test_util.rs index bb57c4cb..e93bc38e 100644 --- a/src/server/test_util.rs +++ b/src/server/test_util.rs @@ -5,7 +5,7 @@ use std::net::{Ipv4Addr, SocketAddr}; use bincode::{Decode, Encode}; use rstest::fixture; -use super::{Bound, WireframeServer}; +use super::WireframeServer; use crate::app::WireframeApp; #[derive(Debug, Clone, PartialEq, Encode, Decode)] @@ -28,7 +28,7 @@ pub fn free_port() -> SocketAddr { .expect("failed to read free port listener address") } -pub fn bind_server(factory: F, addr: SocketAddr) -> WireframeServer +pub fn bind_server(factory: F, addr: SocketAddr) -> WireframeServer where F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, { From 7e90db2b21ef26a38fa0dfb3b35b01f3e59ec91d Mon Sep 17 00:00:00 2001 From: Payton McIntosh Date: Wed, 6 Aug 2025 19:42:12 +0100 Subject: [PATCH 5/6] Adds comprehensive doc examples to ServerConfig and runtime Documents all public builder methods and runtime entrypoints with runnable doctests covering typical usage patterns. Enables consumers to see practical configuration examples without needing external samples. --- src/server/config/mod.rs | 97 ++++++++++++++++++++++++++++++++++++++++ src/server/runtime.rs | 41 +++++++++++++++++ 2 files changed, 138 insertions(+) diff --git a/src/server/config/mod.rs b/src/server/config/mod.rs index dbbaf0aa..954f50a8 100644 --- a/src/server/config/mod.rs +++ b/src/server/config/mod.rs @@ -95,6 +95,15 @@ where } /// Set the number of worker tasks to spawn for the server. + /// + /// # Examples + /// + /// ``` + /// use wireframe::{app::WireframeApp, server::WireframeServer}; + /// + /// let server = WireframeServer::new(|| WireframeApp::default()).workers(4); + /// assert_eq!(server.worker_count(), 4); + /// ``` #[must_use] pub fn workers(mut self, count: usize) -> Self { self.workers = count.max(1); @@ -102,6 +111,24 @@ where } /// Register a callback invoked when the connection preamble decodes successfully. + /// + /// # Examples + /// + /// ``` + /// use std::sync::Arc; + /// + /// use bincode::{Decode, Encode}; + /// use futures::FutureExt; + /// use wireframe::{app::WireframeApp, preamble::Preamble, server::WireframeServer}; + /// + /// #[derive(Encode, Decode)] + /// struct MyPreamble; + /// impl Preamble for MyPreamble {} + /// + /// let server = WireframeServer::new(|| WireframeApp::default()) + /// .with_preamble::() + /// .on_preamble_decode_success(|_preamble: &MyPreamble, _stream| async { Ok(()) }.boxed()); + /// ``` #[must_use] pub fn on_preamble_decode_success(mut self, handler: H) -> Self where @@ -115,6 +142,19 @@ where } /// Register a callback invoked when the connection preamble fails to decode. + /// + /// # Examples + /// + /// ``` + /// use bincode::error::DecodeError; + /// use wireframe::{app::WireframeApp, server::WireframeServer}; + /// + /// let server = WireframeServer::new(|| WireframeApp::default()).on_preamble_decode_failure( + /// |_error: &DecodeError| { + /// eprintln!("Failed to decode preamble"); + /// }, + /// ); + /// ``` #[must_use] pub fn on_preamble_decode_failure(mut self, handler: H) -> Self where @@ -125,6 +165,16 @@ where } /// Configure a channel used to signal when the server is ready to accept connections. + /// + /// # Examples + /// + /// ``` + /// use tokio::sync::oneshot; + /// use wireframe::{app::WireframeApp, server::WireframeServer}; + /// + /// let (tx, _rx) = oneshot::channel(); + /// let server = WireframeServer::new(|| WireframeApp::default()).ready_signal(tx); + /// ``` #[must_use] pub fn ready_signal(mut self, tx: oneshot::Sender<()>) -> Self { self.ready_tx = Some(tx); @@ -132,11 +182,33 @@ where } /// Returns the configured number of worker tasks for the server. + /// + /// # Examples + /// + /// ``` + /// use wireframe::{app::WireframeApp, server::WireframeServer}; + /// + /// let server = WireframeServer::new(|| WireframeApp::default()).workers(8); + /// assert_eq!(server.worker_count(), 8); + /// ``` #[inline] #[must_use] pub const fn worker_count(&self) -> usize { self.workers } /// Returns the bound address, or `None` if not yet bound. + /// + /// # Examples + /// + /// ``` + /// use std::net::{Ipv4Addr, SocketAddr}; + /// + /// use wireframe::{app::WireframeApp, server::WireframeServer}; + /// + /// let server = WireframeServer::new(|| WireframeApp::default()) + /// .bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) + /// .expect("Failed to bind"); + /// assert!(server.local_addr().is_some()); + /// ``` #[must_use] pub fn local_addr(&self) -> Option { self.listener.as_ref().and_then(|l| l.local_addr().ok()) @@ -144,6 +216,18 @@ where /// Bind to a fresh address. /// + /// # Examples + /// + /// ``` + /// use std::net::{Ipv4Addr, SocketAddr}; + /// + /// use wireframe::{app::WireframeApp, server::WireframeServer}; + /// + /// let server = WireframeServer::new(|| WireframeApp::default()) + /// .bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))); + /// assert!(server.is_ok()); + /// ``` + /// /// # Errors /// Returns an `io::Error` if binding or configuring the listener fails. pub fn bind(self, addr: SocketAddr) -> io::Result { @@ -153,6 +237,19 @@ where /// Bind to an existing `StdTcpListener`. /// + /// # Examples + /// + /// ``` + /// use std::net::{Ipv4Addr, SocketAddr, TcpListener as StdTcpListener}; + /// + /// use wireframe::{app::WireframeApp, server::WireframeServer}; + /// + /// let std_listener = StdTcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) + /// .expect("Failed to bind std listener"); + /// let server = WireframeServer::new(|| WireframeApp::default()).bind_listener(std_listener); + /// assert!(server.is_ok()); + /// ``` + /// /// # Errors /// Returns an `io::Error` if configuring the listener fails. pub fn bind_listener(mut self, std: StdTcpListener) -> io::Result { diff --git a/src/server/runtime.rs b/src/server/runtime.rs index c9d81f85..7ae3a02f 100644 --- a/src/server/runtime.rs +++ b/src/server/runtime.rs @@ -31,6 +31,20 @@ where /// /// Spawns the configured number of worker tasks and awaits Ctrl+C for shutdown. /// + /// # Examples + /// + /// ```no_run + /// use wireframe::{app::WireframeApp, server::WireframeServer}; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<(), Box> { + /// let server = + /// WireframeServer::new(|| WireframeApp::default()).bind(([127, 0, 0, 1], 8080).into())?; + /// server.run().await?; + /// # Ok(()) + /// # } + /// ``` + /// /// # Errors /// /// Returns an [`io::Error`] if accepting a connection fails. @@ -43,6 +57,33 @@ where /// Run the server until the `shutdown` future resolves. /// + /// # Examples + /// + /// ``` + /// use tokio::sync::oneshot; + /// use wireframe::{app::WireframeApp, server::WireframeServer}; + /// + /// # #[tokio::main] + /// # async fn main() -> Result<(), Box> { + /// let server = + /// WireframeServer::new(|| WireframeApp::default()).bind(([127, 0, 0, 1], 0).into())?; + /// + /// let (tx, rx) = oneshot::channel::<()>(); + /// let handle = tokio::spawn(async move { + /// server + /// .run_with_shutdown(async { + /// let _ = rx.await; + /// }) + /// .await + /// }); + /// + /// // Signal shutdown + /// let _ = tx.send(()); + /// handle.await??; + /// # Ok(()) + /// # } + /// ``` + /// /// # Errors /// /// Returns an [`io::Error`] if accepting a connection fails during runtime. From 7ffae35e782681dba01e7188fda494b7f20dc74b Mon Sep 17 00:00:00 2001 From: Payton McIntosh Date: Wed, 6 Aug 2025 20:16:05 +0100 Subject: [PATCH 6/6] Refactors server tests for clarity and alignment Extracts repeated default worker count calculation into a dedicated helper Simplifies callback registration tests by removing unnecessary network setup Streamlines assertion logic and removes a redundant test for callback resets Updates variable naming for consistency and drops unused imports --- src/server/config/tests.rs | 85 ++++++++++---------------------------- 1 file changed, 21 insertions(+), 64 deletions(-) diff --git a/src/server/config/tests.rs b/src/server/config/tests.rs index 476d44e5..93eaa978 100644 --- a/src/server/config/tests.rs +++ b/src/server/config/tests.rs @@ -6,7 +6,7 @@ //! cases via `rstest`. use std::{ - net::{Ipv4Addr, SocketAddr}, + net::SocketAddr, sync::{ Arc, atomic::{AtomicUsize, Ordering}, @@ -24,11 +24,15 @@ use crate::server::test_util::{ server_with_preamble, }; +fn expected_default_worker_count() -> usize { + // Mirror the default worker logic to keep tests aligned with `WireframeServer::new`. + std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get) +} + #[rstest] fn test_new_server_creation(factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static) { let server = WireframeServer::new(factory); - assert!(server.worker_count() >= 1); - assert!(server.local_addr().is_none()); + assert!(server.worker_count() >= 1 && server.local_addr().is_none()); } #[rstest] @@ -36,8 +40,7 @@ fn test_new_server_default_worker_count( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { let server = WireframeServer::new(factory); - let expected = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get); - assert_eq!(server.worker_count(), expected); + assert_eq!(server.worker_count(), expected_default_worker_count()); } #[rstest] @@ -55,8 +58,7 @@ fn test_with_preamble_type_conversion( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { let server = WireframeServer::new(factory).with_preamble::(); - let expected = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get); - assert_eq!(server.worker_count(), expected); + assert_eq!(server.worker_count(), expected_default_worker_count()); } #[rstest] @@ -98,6 +100,7 @@ async fn test_preamble_callback_registration( ) { let counter = Arc::new(AtomicUsize::new(0)); let c = counter.clone(); + let server = server_with_preamble(factory); let server = match callback_type { "success" => server.on_preamble_decode_success(move |_p: &TestPreamble, _| { @@ -110,41 +113,15 @@ async fn test_preamble_callback_registration( "failure" => server.on_preamble_decode_failure(move |_err: &DecodeError| { c.fetch_add(1, Ordering::SeqCst); }), - _ => unreachable!("invalid case"), + _ => panic!("Invalid callback type"), }; + assert_eq!(counter.load(Ordering::SeqCst), 0); match callback_type { - "success" => { - let cb = server - .on_preamble_success - .as_ref() - .expect("callback not registered"); - let addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0); - let listener = tokio::net::TcpListener::bind(addr) - .await - .expect("bind temporary listener"); - let addr = listener.local_addr().expect("read listener addr"); - let mut stream = tokio::net::TcpStream::connect(addr) - .await - .expect("connect stream"); - drop(listener); - let preamble = TestPreamble { - id: 0, - message: String::new(), - }; - cb(&preamble, &mut stream).await.expect("callback failed"); - } - "failure" => { - let cb = server - .on_preamble_failure - .as_ref() - .expect("callback not registered"); - cb(&DecodeError::OtherString(String::from("err"))); - } - _ => unreachable!("invalid case"), + "success" => assert!(server.on_preamble_success.is_some()), + "failure" => assert!(server.on_preamble_failure.is_some()), + _ => unreachable!(), } - - assert_eq!(counter.load(Ordering::SeqCst), 1); } #[rstest] @@ -153,13 +130,13 @@ async fn test_method_chaining( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, free_port: SocketAddr, ) { - let counter = Arc::new(AtomicUsize::new(0)); - let c = counter.clone(); + let callback_invoked = Arc::new(AtomicUsize::new(0)); + let counter = callback_invoked.clone(); let server = WireframeServer::new(factory) .workers(2) .with_preamble::() .on_preamble_decode_success(move |_p: &TestPreamble, _| { - let c = c.clone(); + let c = counter.clone(); Box::pin(async move { c.fetch_add(1, Ordering::SeqCst); Ok(()) @@ -170,6 +147,7 @@ async fn test_method_chaining( .expect("Failed to bind"); assert_eq!(server.worker_count(), 2); assert!(server.local_addr().is_some()); + assert_eq!(callback_invoked.load(Ordering::SeqCst), 0); } #[rstest] @@ -186,28 +164,6 @@ async fn test_server_configuration_persistence( assert!(server.local_addr().is_some()); } -#[rstest] -fn test_preamble_callbacks_reset_on_type_change( - factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, -) { - let server = WireframeServer::new(factory) - .on_preamble_decode_success(|&(), _| Box::pin(async { Ok(()) })) - .on_preamble_decode_failure(|_: &DecodeError| {}); - let counter = Arc::new(AtomicUsize::new(0)); - let c = counter.clone(); - let _server = server - .with_preamble::() - .on_preamble_decode_success(move |_p: &TestPreamble, _| { - let c = c.clone(); - Box::pin(async move { - c.fetch_add(1, Ordering::SeqCst); - Ok(()) - }) - }) - .on_preamble_decode_failure(|_: &DecodeError| {}); - assert_eq!(counter.load(Ordering::SeqCst), 0); -} - #[rstest] fn test_extreme_worker_counts(factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static) { let mut server = WireframeServer::new(factory); @@ -228,12 +184,13 @@ async fn test_bind_to_multiple_addresses( let addr2 = listener2 .local_addr() .expect("failed to get second listener address"); + drop(listener2); + let server = WireframeServer::new(factory); let server = server .bind(free_port) .expect("Failed to bind first address"); let first = server.local_addr().expect("first bound address missing"); - drop(listener2); let server = server.bind(addr2).expect("Failed to bind second address"); let second = server.local_addr().expect("second bound address missing"); assert_ne!(first.port(), second.port());