Refactored tcp module

Author: Yuri6037

core/src/tcp/client.rs

@@ -28,9 +28,9 @@
 
 //! A basic TCP client implementation designed for long-running connections.
 
-use crate::tcp::buffer::ChannelBuffer;
-use crate::tcp::util::{DataMsg, Network, ReadyEvent};
-use crate::tcp::{NetReceiver, BYTES_CHANNEL_SIZE};
+use crate::tcp::util::buffer::{ChannelBuffer, NetReceiver};
+use crate::tcp::util::DataMsg;
+use crate::tcp::BYTES_CHANNEL_SIZE;
 use bp3d_debug::{error, trace};
 use std::future::Future;
 use std::sync::Arc;
@@ -39,6 +39,7 @@ use tokio::net::{TcpStream, ToSocketAddrs};
 use tokio::select;
 use tokio::sync::mpsc::error::{SendError, TrySendError};
 use tokio::sync::{mpsc, watch};
+use crate::tcp::util::net::{Network, ReadyEvent};
 use crate::util::barrier;
 
 /// The reader trait which is supposed to handle the actual data reading loop.
@@ -185,7 +186,7 @@ impl<F: Factory> Builder<F> {
                 if let Err(e) = reader.recv(&mut net).await {
                     error!({?net}, "Client error: {}", e);
                 }
-                net.channel_buffer.close();
+                net.close();
             });
             loop {
                 select! {

core/src/tcp/mod.rs

@@ -38,7 +38,3 @@ pub mod client;
 pub mod server;
 
 pub mod util;
-
-mod buffer;
-//TODO: move to util
-pub use buffer::NetReceiver;

core/src/tcp/server/client.rs

@@ -26,15 +26,16 @@
 // NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use crate::tcp::buffer::ChannelBuffer;
-use crate::tcp::util::{DataMsg, Network, ReadyEvent};
-use crate::tcp::{NetReceiver, BYTES_CHANNEL_SIZE};
+use crate::tcp::util::buffer::{ChannelBuffer, NetReceiver};
+use crate::tcp::util::DataMsg;
+use crate::tcp::BYTES_CHANNEL_SIZE;
 use bp3d_debug::{error, trace};
 use std::future::Future;
 use tokio::io::AsyncWriteExt;
 use tokio::select;
 use tokio::sync::mpsc;
 use tokio::sync::watch;
+use crate::tcp::util::net::{Network, ReadyEvent};
 use crate::util::barrier;
 
 /// Represents a client event handler.
@@ -77,7 +78,7 @@ impl<H: Handler + Send + 'static> ClientTask<'_, H> {
             if let Err(e) = self.handler.recv(&mut net).await {
                 error!({?net}, "Client error: {}", e);
             }
-            net.channel_buffer.close();
+            net.close();
             self.handler
         });
         loop {

core/src/tcp/server/server.rs

@@ -27,7 +27,7 @@
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use crate::tcp::server::client::ClientTask;
-use crate::tcp::util::{DataMsg, Network};
+use crate::tcp::util::DataMsg;
 use bp3d_debug::{debug, trace};
 use std::future::Future;
 use std::net::Ipv4Addr;
@@ -41,6 +41,7 @@ use tokio::sync::mpsc;
 use tokio::sync::mpsc::error::{SendError, TrySendError};
 use tokio::sync::watch;
 use tokio::task::JoinSet;
+use crate::tcp::util::net::Network;
 use crate::util::barrier;
 
 /// A factory trait which can be used to create the instance of the main server event handler.

core/src/tcp/buffer.rs renamed to core/src/tcp/util/buffer.rs

@@ -26,6 +26,8 @@
 // NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+//! Async buffered receiver based on mpsc channels.
+
 use crate::tcp::BYTES_BUFFER_SIZE;
 use std::fmt::{Debug, Formatter};
 use std::io::{Error, ErrorKind};
@@ -36,12 +38,21 @@ use std::task::{Context, Poll};
 use tokio::io::{AsyncRead, ReadBuf};
 use tokio::sync::mpsc;
 
+/// A single buffer of bytes.
 pub struct Bytes<const N: usize> {
     bytes: [u8; N],
     size: usize,
 }
 
 impl<const N: usize> Bytes<N> {
+    /// Creates a new owned byte buffer.
+    ///
+    /// # Arguments
+    ///
+    /// * `bytes`: the array of bytes to store.
+    /// * `size`: the number of valid bytes in the buffer.
+    ///
+    /// returns: Bytes<{ N }>
     pub fn new(bytes: [u8; N], size: usize) -> Self {
         Self { bytes, size }
     }
@@ -55,15 +66,24 @@ impl<const N: usize> Deref for Bytes<N> {
     }
 }
 
+/// The main async channel buffer.
 pub struct ChannelBuffer<const N: usize> {
     receiver: mpsc::Receiver<Bytes<N>>,
 }
 
 impl<const N: usize> ChannelBuffer<N> {
+    /// Creates a new [ChannelBuffer] by wrapping a mpsc channel.
+    ///
+    /// # Arguments
+    ///
+    /// * `receiver`: the channel to wrap.
+    ///
+    /// returns: ChannelBuffer<{ N }>
     pub fn new(receiver: mpsc::Receiver<Bytes<N>>) -> Self {
         Self { receiver }
     }
 
+    /// Closes this [ChannelBuffer].
     pub fn close(mut self) {
         self.receiver.close();
     }
@@ -92,9 +112,9 @@ impl<const N: usize> AsyncRead for ChannelBuffer<N> {
     }
 }
 
-/// Represents a network receiver.
+/// Represents a network receiver which wraps a [ChannelBuffer].
 pub struct NetReceiver {
-    pub(super) channel_buffer: ChannelBuffer<BYTES_BUFFER_SIZE>,
+    channel_buffer: ChannelBuffer<BYTES_BUFFER_SIZE>,
     addr: SocketAddr,
     id: usize,
 }
@@ -140,6 +160,11 @@ impl NetReceiver {
     pub fn id(&self) -> usize {
         self.id
     }
+
+    /// Closes this [ChannelBuffer].
+    pub fn close(self) {
+        self.channel_buffer.close();
+    }
 }
 
 impl AsyncRead for NetReceiver {

core/src/tcp/util/mod.rs

@@ -0,0 +1,44 @@
+// Copyright (c) 2025, BlockProject 3D
+//
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without modification,
+// are permitted provided that the following conditions are met:
+//
+//     * Redistributions of source code must retain the above copyright notice,
+//       this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above copyright notice,
+//       this list of conditions and the following disclaimer in the documentation
+//       and/or other materials provided with the distribution.
+//     * Neither the name of BlockProject 3D nor the names of its contributors
+//       may be used to endorse or promote products derived from this software
+//       without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+//! Utility module for TCP client or server.
+
+pub mod net;
+pub mod buffer;
+
+use std::fmt::Debug;
+
+#[derive(Clone, Debug)]
+pub(super) struct DataMsg {
+    pub(super) buffer: *const u8,
+    pub(super) buffer_size: usize,
+    #[allow(dead_code)]
+    pub(super) net_id: usize,
+}
+
+unsafe impl Send for DataMsg {}

core/src/tcp/util.rs renamed to core/src/tcp/util/net.rs

@@ -26,29 +26,19 @@
 // NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-//! Utility module for TCP client or server.
+//! TCP network stream async reader/writer tools.
 
-use bp3d_debug::warning;
 use std::fmt::{Debug, Formatter};
 use std::io::{Error, ErrorKind, IoSlice};
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use bp3d_debug::warning;
 use tokio::io::{AsyncRead, AsyncWrite, Interest, ReadBuf};
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::net::TcpStream;
 use tokio::sync::mpsc;
-use crate::tcp::buffer::Bytes;
-
-#[derive(Clone, Debug)]
-pub(super) struct DataMsg {
-    pub(super) buffer: *const u8,
-    pub(super) buffer_size: usize,
-    #[allow(dead_code)]
-    pub(super) net_id: usize,
-}
-
-unsafe impl Send for DataMsg {}
+use crate::tcp::util::buffer::Bytes;
 
 /// The event returned by the ready function in [Network].
 pub enum ReadyEvent {

testprog/src/client.rs

@@ -27,10 +27,10 @@
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use bp3d_net::tcp::client::{Client, Factory, Handler, Reader};
-use bp3d_net::tcp::util::Network;
-use bp3d_net::tcp::NetReceiver;
 use std::sync::Arc;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use bp3d_net::tcp::util::buffer::NetReceiver;
+use bp3d_net::tcp::util::net::Network;
 
 #[derive(Clone)]
 pub struct EchoClient {

testprog/src/server.rs

@@ -28,10 +28,10 @@
 
 use bp3d_debug::debug;
 use bp3d_net::tcp::server::{ClientHandler, Factory, Handler, Server};
-use bp3d_net::tcp::util::Network;
-use bp3d_net::tcp::NetReceiver;
 use std::sync::Arc;
 use tokio::io::{AsyncBufReadExt, BufReader};
+use bp3d_net::tcp::util::buffer::NetReceiver;
+use bp3d_net::tcp::util::net::Network;
 
 pub struct EchoServer {
     server: Arc<Server<EchoServer>>,