Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,7 @@ serial_test = "4.0.1"
metrics-util = "0.20.0"
tracing-test = { version = "0.2.5", features = ["no-env-filter"] }
mockall = "0.15.0"
mockable = "3.0.0"
monotony = { version = "1.0.0", features = ["test-util"] }
criterion = "0.8.2"
cap-std = { version = "4.0.2", features = ["fs_utf8"] }
Expand Down
5 changes: 4 additions & 1 deletion examples/client_echo_login.rs
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,10 @@ type Client = WireframeClient<BincodeSerializer, RewindStream<TcpStream>, ()>;

/// Installs the default formatter once, leaving embedding applications free to
/// provide their own subscriber when this example is reused.
fn init_tracing() { let _ = tracing_subscriber::fmt::try_init(); }
fn init_tracing() {
// A host embedding this example may already own the global subscriber.
let _subscriber_result = tracing_subscriber::fmt::try_init();
}

/// Returns the address used by the companion echo server.
fn server_addr() -> ExampleResult<SocketAddr> { Ok("127.0.0.1:7878".parse()?) }
Expand Down
5 changes: 3 additions & 2 deletions examples/metadata_routing.rs
Original file line number Diff line number Diff line change
Expand Up @@ -108,9 +108,10 @@ async fn run() -> io::Result<()> {
.map_err(|error| io::Error::other(error.to_string()))?;

let mut codec = app.length_codec();
let app = app.prepare().await.map_err(io::Error::other)?;
let prepared_app = app.prepare().await.map_err(io::Error::other)?;
let (mut client, server) = duplex(1024);
let server_task = tokio::spawn(async move { app.handle_connection_result(server).await });
let server_task =
tokio::spawn(async move { prepared_app.handle_connection_result(server).await });

let payload = Ping.to_bytes().map_err(io::Error::other)?;
let mut frame = Vec::new();
Expand Down
9 changes: 6 additions & 3 deletions examples/multi_packet.rs
Original file line number Diff line number Diff line change
Expand Up @@ -93,9 +93,11 @@ fn multi_packet_response() -> Response<Frame> {
}
});

tokio::spawn(async move {
let _summary_task = tokio::spawn(async move {
// Wait for all chunks to be sent before delivering the summary.
let _ = chunk_task.await;
if let Err(error) = chunk_task.await {
tracing::error!(%error, "chunk producer failed");
}
let summary = Frame::summary(TRANSCRIPT.len());
if summary_sender.send(summary).await.is_err() {
tracing::trace!("connection dropped, summary not sent");
Expand All @@ -114,7 +116,8 @@ fn log_frame(frame: &Frame) {
/// Stream errors are propagated so the example demonstrates its normal error
/// boundary rather than silently truncating a multipart response.
async fn run() -> Result<(), WireframeError> {
let _ = tracing_subscriber::fmt::try_init();
// A host embedding this example may already own the global subscriber.
let _subscriber_result = tracing_subscriber::fmt::try_init();
let response = multi_packet_response();
let mut stream = response.into_stream();
while let Some(frame) = stream.try_next().await? {
Expand Down
48 changes: 44 additions & 4 deletions examples/packet_enum.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
use std::{collections::HashMap, future::Future, net::SocketAddr, pin::Pin, sync::Arc};

use async_trait::async_trait;
use mockable::{DefaultEnv, Env};
use tokio::signal;
use tracing::{info, warn};
use wireframe::{
Expand Down Expand Up @@ -104,9 +105,9 @@ fn build_app() -> wireframe::app::Result<App> {
.route(1, Arc::new(handle_packet))
}

/// Read and validate the listener address, reporting malformed configuration.
fn parse_server_addr() -> std::io::Result<SocketAddr> {
let addr_str = std::env::var("SERVER_ADDR").unwrap_or_else(|_| DEFAULT_ADDR.to_string());
/// Validate a listener address, falling back when the environment lookup fails.
fn parse_server_addr(addr: &Result<String, std::env::VarError>) -> std::io::Result<SocketAddr> {
let addr_str = addr.as_deref().unwrap_or(DEFAULT_ADDR);
addr_str.parse().map_err(|error| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
Expand All @@ -120,7 +121,7 @@ async fn run() -> std::io::Result<()> {
let _ = tracing_subscriber::fmt::try_init();
let app = build_app().map_err(|error| std::io::Error::other(error.to_string()))?;
let server = WireframeServer::from_app(app)
.bind(parse_server_addr()?)
.bind(parse_server_addr(&DefaultEnv.raw("SERVER_ADDR"))?)
.map_err(|error| std::io::Error::other(error.to_string()))?;
server
.run_with_shutdown(async {
Expand All @@ -139,3 +140,42 @@ fn main() -> std::io::Result<()> {
.build()?
.block_on(run())
}

#[cfg(test)]
mod tests {
//! Verifies listener address parsing without mutating the process environment.

use std::{env::VarError, ffi::OsString, io, net::SocketAddr};

use rstest::rstest;

use super::{DEFAULT_ADDR, parse_server_addr};

#[rstest]
#[case::missing(VarError::NotPresent)]
#[case::non_unicode(VarError::NotUnicode(OsString::from("raw value")))]
fn uses_default_when_lookup_fails(#[case] error: VarError) -> Result<(), io::Error> {
let addr = parse_server_addr(&Err(error))?;
assert_eq!(addr.to_string(), DEFAULT_ADDR);
Ok(())
}

#[test]
fn accepts_configured_address() -> Result<(), io::Error> {
let addr = parse_server_addr(&Ok("127.0.0.1:9000".to_string()))?;
assert_eq!(addr, SocketAddr::from(([127, 0, 0, 1], 9000)));
Ok(())
}

#[test]
fn rejects_malformed_configured_address() -> Result<(), io::Error> {
let error = parse_server_addr(&Ok("not-an-address".to_string()))
.expect_err("malformed configured address must fail");
assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
assert_eq!(
error.to_string(),
"SERVER_ADDR must be a valid socket address: invalid socket address syntax"
);
Ok(())
}
}
4 changes: 2 additions & 2 deletions examples/resp_codec_impl/codec.rs
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ pub struct RespFrameCodec {
impl RespFrameCodec {
/// Create a new RESP codec with the given maximum frame length.
#[must_use]
pub fn new(max_frame_length: usize) -> Self { Self { max_frame_length } }
pub const fn new(max_frame_length: usize) -> Self { Self { max_frame_length } }
}

/// Tokio codec adapter for RESP frames.
Expand All @@ -38,7 +38,7 @@ pub struct RespAdapter {

impl RespAdapter {
/// Build a Tokio adapter carrying the frame-size policy for one direction.
fn new(max_frame_length: usize) -> Self { Self { max_frame_length } }
const fn new(max_frame_length: usize) -> Self { Self { max_frame_length } }
}

impl Decoder for RespAdapter {
Expand Down
4 changes: 2 additions & 2 deletions examples/resp_codec_impl/encode.rs
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ fn encode_frame_at(frame: &RespFrame, dst: &mut BytesMut, depth: usize) -> Resul
}

/// Count decimal digits without allocating a temporary representation.
fn digits_len_usize(mut value: usize) -> usize {
const fn digits_len_usize(mut value: usize) -> usize {
let mut digits = 1;
while value >= 10 {
value = value.div_euclid(10);
Expand All @@ -121,7 +121,7 @@ fn digits_len_usize(mut value: usize) -> usize {
}

/// Count digits in a signed integer, including its possible minus sign.
fn digits_len_i64(value: i64) -> usize {
const fn digits_len_i64(value: i64) -> usize {
let mut digits = 1;
let mut cursor = value.unsigned_abs();
while cursor >= 10 {
Expand Down
16 changes: 8 additions & 8 deletions examples/resp_codec_impl/parse.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ const MAX_ARRAY_ELEMENTS: usize = 1024;
const MAX_RECURSION_DEPTH: usize = 64;

/// Count the number of decimal digits required to represent a value.
fn decimal_digits(mut value: usize) -> usize {
const fn decimal_digits(mut value: usize) -> usize {
let mut digits = 1;
while value >= 10 {
value = value.div_euclid(10);
Expand All @@ -71,7 +71,7 @@ struct ParseContext<'a> {

impl<'a> ParseContext<'a> {
/// Capture the immutable buffer view and limits for one parser operation.
fn new(buf: &'a BytesMut, start: usize, max_frame_length: usize) -> Self {
const fn new(buf: &'a BytesMut, start: usize, max_frame_length: usize) -> Self {
Self {
buf,
start,
Expand Down Expand Up @@ -223,9 +223,9 @@ fn parse_bulk_length(
return Ok(Some((BulkLength::Null, next)));
}

let len = usize::try_from(len)
let payload_length = usize::try_from(len)
.map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "bulk length too large"))?;
Ok(Some((BulkLength::Sized(len), next)))
Ok(Some((BulkLength::Sized(payload_length), next)))
}

/// Validate a bulk payload without consuming incomplete input.
Expand Down Expand Up @@ -307,19 +307,19 @@ fn parse_array(
"invalid negative array length",
));
}
let count = usize::try_from(count)
let element_count = usize::try_from(count)
.map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "array too large"))?;
// Limit element count to prevent allocation-based DoS. The cumulative
// byte validation at the end of this function enforces the actual
// byte-budget constraint.
if count > MAX_ARRAY_ELEMENTS {
if element_count > MAX_ARRAY_ELEMENTS {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"array element count exceeds limit",
));
}
let mut frames = Vec::with_capacity(count);
for _ in 0..count {
let mut frames = Vec::with_capacity(element_count);
for _ in 0..element_count {
let Some((frame, consumed)) = parse_frame_at(buf, cursor, max_frame_length, depth + 1)?
else {
return Ok(None);
Expand Down
Loading