switchy_tcp/
lib.rs

1//! Generic TCP stream and listener abstractions for async Rust.
2//!
3//! This crate provides generic traits and implementations for TCP networking that work
4//! across different async runtimes. It supports both real tokio-based networking and an
5//! in-memory simulator for testing.
6//!
7//! # Features
8//!
9//! * `tokio` - Real TCP networking using tokio
10//! * `simulator` - In-memory TCP simulator for testing without actual network I/O
11//!
12//! # Examples
13//!
14//! ```rust,no_run
15//! # #[cfg(feature = "tokio")]
16//! # {
17//! use switchy_tcp::{TokioTcpListener, GenericTcpListener};
18//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
19//! // Create a TCP listener
20//! let listener = TokioTcpListener::bind("127.0.0.1:8080").await?;
21//!
22//! // Accept incoming connections
23//! let (stream, addr) = listener.accept().await?;
24//! println!("Connection from: {}", addr);
25//! # Ok(())
26//! # }
27//! # }
28//! ```
29
30#![cfg_attr(feature = "fail-on-warnings", deny(warnings))]
31#![warn(clippy::all, clippy::pedantic, clippy::nursery, clippy::cargo)]
32#![allow(clippy::multiple_crate_versions)]
33
34use std::{marker::PhantomData, net::SocketAddr};
35
36use ::tokio::io::{AsyncRead, AsyncWrite};
37use async_trait::async_trait;
38use thiserror::Error;
39
40/// Real TCP networking implementation using tokio.
41///
42/// This module provides TCP streams and listeners backed by the tokio runtime for actual
43/// network I/O operations.
44#[cfg(feature = "tokio")]
45pub mod tokio;
46
47/// In-memory TCP simulator for testing.
48///
49/// This module provides TCP streams and listeners that simulate network behavior in-memory
50/// without actual network I/O. Useful for deterministic testing and avoiding port conflicts.
51#[cfg(feature = "simulator")]
52pub mod simulator;
53
54/// Error types for TCP operations.
55#[derive(Debug, Error)]
56pub enum Error {
57    /// I/O error from the underlying stream or listener.
58    #[error(transparent)]
59    IO(#[from] ::std::io::Error),
60    /// Failed to parse a socket address.
61    #[error(transparent)]
62    AddrParse(#[from] ::std::net::AddrParseError),
63    /// Failed to parse an integer (typically a port number).
64    #[error(transparent)]
65    ParseInt(#[from] std::num::ParseIntError),
66    /// Failed to send data over a channel (simulator only).
67    #[cfg(feature = "simulator")]
68    #[error("Send error")]
69    Send,
70}
71
72/// Generic trait for TCP listeners that can accept connections.
73///
74/// # Errors
75///
76/// * `accept` may fail if the underlying listener encounters an error while accepting a connection
77#[async_trait]
78pub trait GenericTcpListener<T>: Send + Sync {
79    /// Accepts a new incoming connection.
80    ///
81    /// Returns the connected stream and the remote address.
82    ///
83    /// # Errors
84    ///
85    /// * If the underlying listener fails to accept a connection
86    async fn accept(&self) -> Result<(T, SocketAddr), Error>;
87}
88
89/// Generic trait for TCP streams that can be split into read and write halves.
90///
91/// Provides methods for splitting a stream into separate read and write halves, and for
92/// querying the local and remote addresses of the connection.
93pub trait GenericTcpStream<R: GenericTcpStreamReadHalf, W: GenericTcpStreamWriteHalf>:
94    AsyncRead + AsyncWrite + Send + Sync + Unpin
95{
96    /// Splits the stream into separate read and write halves.
97    fn into_split(self) -> (R, W);
98
99    /// Returns the local address of this stream.
100    ///
101    /// # Errors
102    ///
103    /// * If the underlying `TcpStream` fails to get the `local_addr`
104    fn local_addr(&self) -> std::io::Result<SocketAddr>;
105
106    /// Returns the remote address of this stream.
107    ///
108    /// # Errors
109    ///
110    /// * If the underlying `TcpStream` fails to get the `peer_addr`
111    fn peer_addr(&self) -> std::io::Result<SocketAddr>;
112}
113
114/// Generic trait for the read half of a TCP stream.
115///
116/// This trait marks types that represent the readable half of a split TCP stream.
117pub trait GenericTcpStreamReadHalf: AsyncRead + Send + Sync + Unpin {}
118
119/// Generic trait for the write half of a TCP stream.
120///
121/// This trait marks types that represent the writable half of a split TCP stream.
122pub trait GenericTcpStreamWriteHalf: AsyncWrite + Send + Sync + Unpin {}
123
124/// Wrapper type for generic TCP listeners.
125///
126/// This type wraps implementations of `GenericTcpListener` and provides a unified interface
127/// for accepting TCP connections. It is typically instantiated via type aliases like
128/// `TokioTcpListener` or `SimulatorTcpListener`.
129pub struct TcpListenerWrapper<
130    R: GenericTcpStreamReadHalf,
131    W: GenericTcpStreamWriteHalf,
132    S: GenericTcpStream<R, W>,
133    T: GenericTcpListener<S>,
134>(T, PhantomData<R>, PhantomData<W>, PhantomData<S>);
135
136/// Wrapper type for generic TCP streams.
137///
138/// This type wraps implementations of `GenericTcpStream` and provides a unified interface
139/// for reading and writing over TCP connections. It is typically instantiated via type aliases
140/// like `TokioTcpStream` or `SimulatorTcpStream`.
141pub struct TcpStreamWrapper<
142    R: GenericTcpStreamReadHalf,
143    W: GenericTcpStreamWriteHalf,
144    T: GenericTcpStream<R, W>,
145>(T, PhantomData<R>, PhantomData<W>);
146
147#[allow(unused)]
148macro_rules! impl_http {
149    ($module:ident, $local_module:ident $(,)?) => {
150        paste::paste! {
151            pub use [< impl_ $module >]::*;
152        }
153
154        mod $local_module {
155            use std::pin::pin;
156
157            use crate::*;
158
159            paste::paste! {
160                #[doc = concat!("Read half of a ", stringify!($module), " TCP stream.\n\nWraps the underlying read half to provide a generic interface.")]
161                pub type [< $module:camel TcpStreamReadHalf >] = $module::TcpStreamReadHalf;
162                type ModuleTcpStreamReadHalf = [< $module:camel TcpStreamReadHalf >];
163
164                #[doc = concat!("Write half of a ", stringify!($module), " TCP stream.\n\nWraps the underlying write half to provide a generic interface.")]
165                pub type [< $module:camel TcpStreamWriteHalf >] = $module::TcpStreamWriteHalf;
166                type ModuleTcpStreamWriteHalf = [< $module:camel TcpStreamWriteHalf >];
167
168                #[doc = concat!("TCP stream for ", stringify!($module), ".\n\nWraps the underlying stream to provide a generic interface that can be split into read and write halves.")]
169                pub type [< $module:camel TcpStream >] = TcpStreamWrapper<ModuleTcpStreamReadHalf, ModuleTcpStreamWriteHalf, $module::TcpStream>;
170                type ModuleTcpStream = [< $module:camel TcpStream >];
171
172                #[doc = concat!("TCP listener for ", stringify!($module), ".\n\nWraps the underlying listener to provide a generic interface for accepting connections.")]
173                pub type [< $module:camel TcpListener >] = TcpListenerWrapper<ModuleTcpStreamReadHalf, ModuleTcpStreamWriteHalf, ModuleTcpStream, $module::TcpListener>;
174                type ModuleTcpListener = [< $module:camel TcpListener >];
175            }
176
177            #[async_trait]
178            impl GenericTcpListener<ModuleTcpStream> for ModuleTcpListener {
179                async fn accept(&self) -> Result<(ModuleTcpStream, SocketAddr), Error> {
180                    self.0.accept().await
181                }
182            }
183
184            impl GenericTcpStream<ModuleTcpStreamReadHalf, ModuleTcpStreamWriteHalf> for ModuleTcpStream {
185                fn into_split(self) -> (ModuleTcpStreamReadHalf, ModuleTcpStreamWriteHalf) {
186                    self.0.into_split()
187                }
188
189                fn local_addr(&self) -> std::io::Result<SocketAddr> {
190                    self.0.local_addr()
191                }
192
193                fn peer_addr(&self) -> std::io::Result<SocketAddr> {
194                    self.0.peer_addr()
195                }
196            }
197
198            impl ModuleTcpStream {
199                /// Connects to a remote TCP server at the specified address.
200                ///
201                /// # Errors
202                ///
203                /// * If the underlying `TcpStream` fails to connect
204                pub async fn connect(addr: &str) -> std::io::Result<Self> {
205                    Ok(Self($module::TcpStream::connect(addr).await?, PhantomData, PhantomData))
206                }
207
208                /// Returns the local socket address of this stream.
209                ///
210                /// # Errors
211                ///
212                /// * If the underlying `TcpStream` fails to get the `local_addr`
213                pub fn local_addr(&self) -> std::io::Result<SocketAddr> {
214                    self.0.local_addr()
215                }
216
217                /// Returns the remote peer socket address of this stream.
218                ///
219                /// # Errors
220                ///
221                /// * If the underlying `TcpStream` fails to get the `peer_addr`
222                pub fn peer_addr(&self) -> std::io::Result<SocketAddr> {
223                    self.0.peer_addr()
224                }
225            }
226
227            impl AsyncRead for ModuleTcpStream {
228                fn poll_read(
229                    self: std::pin::Pin<&mut Self>,
230                    cx: &mut std::task::Context<'_>,
231                    buf: &mut ::tokio::io::ReadBuf<'_>,
232                ) -> std::task::Poll<std::io::Result<()>> {
233                    let this = self.get_mut();
234                    let inner = &mut this.0;
235                    let inner = pin!(inner);
236                    AsyncRead::poll_read(inner, cx, buf)
237                }
238            }
239
240            impl AsyncWrite for ModuleTcpStream {
241                fn poll_write(
242                    self: std::pin::Pin<&mut Self>,
243                    cx: &mut std::task::Context<'_>,
244                    buf: &[u8],
245                ) -> std::task::Poll<Result<usize, std::io::Error>> {
246                    let this = self.get_mut();
247                    let inner = &mut this.0;
248                    let inner = pin!(inner);
249                    AsyncWrite::poll_write(inner, cx, buf)
250                }
251
252                fn poll_flush(
253                    self: std::pin::Pin<&mut Self>,
254                    cx: &mut std::task::Context<'_>,
255                ) -> std::task::Poll<Result<(), std::io::Error>> {
256                    let this = self.get_mut();
257                    let inner = &mut this.0;
258                    let inner = pin!(inner);
259                    AsyncWrite::poll_flush(inner, cx)
260                }
261
262                fn poll_shutdown(
263                    self: std::pin::Pin<&mut Self>,
264                    cx: &mut std::task::Context<'_>,
265                ) -> std::task::Poll<Result<(), std::io::Error>> {
266                    let this = self.get_mut();
267                    let inner = &mut this.0;
268                    let inner = pin!(inner);
269                    AsyncWrite::poll_shutdown(inner, cx)
270                }
271            }
272        }
273    };
274}
275
276#[cfg(feature = "simulator")]
277impl_http!(simulator, impl_simulator);
278
279#[cfg(feature = "tokio")]
280impl_http!(tokio, impl_tokio);
281
282#[allow(unused)]
283macro_rules! impl_gen_types {
284    ($module:ident $(,)?) => {
285        paste::paste! {
286            /// Default TCP listener type for the current feature configuration.
287            ///
288            /// This type alias points to the appropriate listener implementation based on
289            /// enabled features. With the `simulator` feature, it uses the in-memory simulator.
290            /// Otherwise, it uses the tokio-based implementation.
291            pub type TcpListener = [< $module:camel TcpListener >];
292
293            /// Default TCP stream type for the current feature configuration.
294            ///
295            /// This type alias points to the appropriate stream implementation based on
296            /// enabled features. With the `simulator` feature, it uses the in-memory simulator.
297            /// Otherwise, it uses the tokio-based implementation.
298            pub type TcpStream = [< $module:camel TcpStream >];
299
300            /// Default TCP stream read half type for the current feature configuration.
301            ///
302            /// This type alias points to the appropriate read half implementation based on
303            /// enabled features. With the `simulator` feature, it uses the in-memory simulator.
304            /// Otherwise, it uses the tokio-based implementation.
305            pub type TcpStreamReadHalf = [< $module:camel TcpStreamReadHalf >];
306
307            /// Default TCP stream write half type for the current feature configuration.
308            ///
309            /// This type alias points to the appropriate write half implementation based on
310            /// enabled features. With the `simulator` feature, it uses the in-memory simulator.
311            /// Otherwise, it uses the tokio-based implementation.
312            pub type TcpStreamWriteHalf = [< $module:camel TcpStreamWriteHalf >];
313        }
314    };
315}
316
317#[cfg(feature = "simulator")]
318impl_gen_types!(simulator);
319
320#[cfg(all(not(feature = "simulator"), feature = "tokio"))]
321impl_gen_types!(tokio);
322
323#[allow(unused)]
324macro_rules! impl_read_inner {
325    ($type:ty $(,)?) => {
326        impl tokio::io::AsyncRead for $type {
327            fn poll_read(
328                self: std::pin::Pin<&mut Self>,
329                cx: &mut std::task::Context<'_>,
330                buf: &mut ::tokio::io::ReadBuf<'_>,
331            ) -> std::task::Poll<std::io::Result<()>> {
332                let this = self.get_mut();
333                let inner = &mut this.0;
334                let inner = std::pin::pin!(inner);
335                tokio::io::AsyncRead::poll_read(inner, cx, buf)
336            }
337        }
338    };
339}
340
341#[allow(unused)]
342macro_rules! impl_write_inner {
343    ($type:ty $(,)?) => {
344        impl tokio::io::AsyncWrite for $type {
345            fn poll_write(
346                self: std::pin::Pin<&mut Self>,
347                cx: &mut std::task::Context<'_>,
348                buf: &[u8],
349            ) -> std::task::Poll<Result<usize, std::io::Error>> {
350                let this = self.get_mut();
351                let inner = &mut this.0;
352                let inner = std::pin::pin!(inner);
353                tokio::io::AsyncWrite::poll_write(inner, cx, buf)
354            }
355
356            fn poll_flush(
357                self: std::pin::Pin<&mut Self>,
358                cx: &mut std::task::Context<'_>,
359            ) -> std::task::Poll<Result<(), std::io::Error>> {
360                let this = self.get_mut();
361                let inner = &mut this.0;
362                let inner = std::pin::pin!(inner);
363                tokio::io::AsyncWrite::poll_flush(inner, cx)
364            }
365
366            fn poll_shutdown(
367                self: std::pin::Pin<&mut Self>,
368                cx: &mut std::task::Context<'_>,
369            ) -> std::task::Poll<Result<(), std::io::Error>> {
370                let this = self.get_mut();
371                let inner = &mut this.0;
372                let inner = std::pin::pin!(inner);
373                tokio::io::AsyncWrite::poll_shutdown(inner, cx)
374            }
375        }
376    };
377}
378
379#[allow(unused)]
380pub(crate) use impl_read_inner;
381#[allow(unused)]
382pub(crate) use impl_write_inner;
switchy_tcp/lib.rs

switchy_tcp/
lib.rs
