src/connections.rs

@@ -6,14 +6,12 @@ use std::{
     time::{Duration, Instant},
 };
 
-use serde::{Serialize, de::DeserializeOwned};
 use thiserror::Error;
 
 use crate::{
     PeerConfig, PeerMessage,
-    listener::{DATAGRAM_SIZE, ListenerError},
+    listener::{DATAGRAM_SIZE, Datagram, ListenerError},
-    package::{Message, Messages, Package},
+    package::Package,
-    stats::NetStats,
 };
 
 /// Represents an error during sending a package for any number of reason
@@ -41,29 +39,22 @@ pub enum ConnectionError {
 }
 
 /// Manages connections for the Peer
-pub(crate) struct ConnectionManager<
+pub(crate) struct ConnectionManager {
-    T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static,
-> {
     /// Wraps UdpSocket with some helper methods
     pub udp: UdpWrapper,
-    connections: HashMap<SocketAddr, Connection<T>>,
+    connections: HashMap<SocketAddr, Connection>,
     closing_since: Option<Instant>,
     config: PeerConfig,
-
-    accepting_connections: bool,
 }
 
-impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static>
+impl ConnectionManager {
-    ConnectionManager<T>
-{
     /// Creates a new ConnectionManager with the given socket and config
-    pub fn new(socket: Arc<UdpSocket>, config: PeerConfig) -> ConnectionManager<T> {
+    pub fn new(socket: Arc<UdpSocket>, config: PeerConfig) -> ConnectionManager {
         ConnectionManager {
             udp: UdpWrapper::new(socket),
             connections: HashMap::new(),
             closing_since: None,
             config,
-            accepting_connections: true,
         }
     }
 
@@ -103,9 +94,8 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
                 if duration > self.config.ping_interval {
                     match conn.state {
                         ConnectionState::ReceivingConnection | ConnectionState::Connecting => {
-                            match self.udp.send_to(*addr, Package::<T>::Hello) {
+                            match self.udp.send_to(*addr, Package::Hello) {
-                                Ok(bytes) => {
+                                Ok(_) => {
-                                    conn.bytes_tx += bytes;
                                     conn.last_sent_ping = now;
                                 }
                                 Err(err) => {
@@ -116,9 +106,8 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
                             }
                         }
                         ConnectionState::Connected | ConnectionState::ConnectingNearlyReady => {
-                            match self.udp.send_to(*addr, Package::<T>::Ping) {
+                            match self.udp.send_to(*addr, Package::Ping) {
-                                Ok(bytes) => {
+                                Ok(_) => {
-                                    conn.bytes_tx += bytes;
                                     conn.last_sent_ping = now;
                                 }
                                 Err(err) => {
@@ -129,9 +118,8 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
                             }
                         }
                         ConnectionState::Closing | ConnectionState::ReceivingClosing => {
-                            match self.udp.send_to(*addr, Package::<T>::Close) {
+                            match self.udp.send_to(*addr, Package::Close) {
-                                Ok(bytes) => {
+                                Ok(_) => {
-                                    conn.bytes_tx += bytes;
                                     conn.last_sent_ping = now;
                                 }
                                 Err(err) => {
@@ -148,77 +136,6 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
         }
     }
 
-    pub fn send_queued_messages(&mut self) {
-        let now = Instant::now();
-        for (addr, conn) in &mut self.connections {
-            if now - conn.last_msg_send > self.config.message_retry {
-                let messages = Messages {
-                    ack: conn.ack,
-                    reliable: conn.reliable_queue.clone(),
-                    unreliable: Vec::new(),
-                    messages_sent: conn.messages_sent + 1,
-                };
-                match self.udp.send_to(*addr, Package::Messages(messages)) {
-                    Ok(bytes) => {
-                        conn.bytes_tx += bytes;
-                        conn.last_msg_send = now;
-                        conn.messages_sent += 1;
-                    }
-                    Err(err) => {
-                        conn.closing_since = Instant::now();
-                        conn.state = ConnectionState::Error;
-                        conn.error = Some(ConnectionError::SendError(err));
-                    }
-                }
-            }
-        }
-    }
-
-    pub fn send(&mut self, addr: &SocketAddr, message: T, reliable: bool) {
-        if let Some(conn) = self.connections.get_mut(addr) {
-            if conn.state == ConnectionState::Connected {
-                let message_id = conn.message_counter;
-                if reliable {
-                    // Only increment message counter for reliable messages
-                    conn.message_counter += 1;
-                }
-
-                let message = Message {
-                    message_id,
-                    message,
-                };
-
-                let mut reliables = conn.reliable_queue.clone();
-                let mut unreliables = Vec::new();
-                if reliable {
-                    reliables.push(message.clone());
-                    conn.reliable_queue.push(message);
-                } else {
-                    unreliables.push(message);
-                }
-
-                let messages = Messages {
-                    ack: conn.ack,
-                    reliable: reliables,
-                    unreliable: unreliables,
-                    messages_sent: conn.messages_sent + 1,
-                };
-                match self.udp.send_to(*addr, Package::Messages(messages)) {
-                    Ok(bytes) => {
-                        conn.bytes_tx += bytes;
-                        conn.last_msg_send = Instant::now();
-                        conn.messages_sent += 1;
-                    }
-                    Err(err) => {
-                        conn.closing_since = Instant::now();
-                        conn.state = ConnectionState::Error;
-                        conn.error = Some(ConnectionError::SendError(err));
-                    }
-                }
-            }
-        }
-    }
-
     /// Close the connection to another peer at the given address
     pub fn close_connection(&mut self, addr: &SocketAddr) {
         if let Some(conn) = self.connections.get_mut(addr) {
@@ -242,16 +159,7 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
 
     /// Handle Package from a remote peer
     #[must_use]
-    pub fn handle_package(
+    pub fn handle_package(&mut self, package: Package, addr: &SocketAddr) -> Vec<PeerMessage> {
-        &mut self,
-        package: Package<T>,
-        addr: &SocketAddr,
-        bytes: usize,
-    ) -> Vec<PeerMessage<T>> {
-        if let Some(conn) = self.connections.get_mut(addr) {
-            conn.bytes_rx += bytes;
-        }
-
         let mut messages = Vec::new();
 
         match package {
@@ -260,10 +168,6 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
                     if conn.state == ConnectionState::Connecting {
                         conn.state = ConnectionState::ConnectingNearlyReady;
                     }
-                } else {
-                    if !self.accepting_connections {
-                        self.connections
-                            .insert(*addr, Connection::from(*addr, ConnectionState::Closing));
                 } else {
                     self.connections.insert(
                         *addr,
@@ -271,7 +175,6 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
                     );
                 }
             }
-            }
             Package::Ping => {
                 if let Some(conn) = self.connections.get_mut(addr) {
                     if conn.state == ConnectionState::ConnectingNearlyReady
@@ -283,10 +186,8 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
 
                     if conn.state == ConnectionState::Connected {
                         conn.last_recv_ping = Instant::now();
-                        match self.udp.send_to(*addr, Package::<T>::Pong) {
+                        match self.udp.send_to(*addr, Package::Pong) {
-                            Ok(bytes) => {
+                            Ok(_) => {}
-                                conn.bytes_tx += bytes;
-                            }
                             Err(err) => {
                                 conn.closing_since = Instant::now();
                                 conn.state = ConnectionState::Error;
@@ -329,51 +230,6 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
                     self.connections.remove(addr);
                 }
             }
-            Package::Messages(msgs) => {
-                if let Some(conn) = self.connections.get_mut(addr) {
-                    if conn.state == ConnectionState::Connected {
-                        // Store the expected number of messages to have received
-                        // from the given connection
-                        conn.messages_expected = msgs.messages_sent.max(conn.messages_expected);
-                        conn.messages_received += 1;
-
-                        // Clear all messages from the queue that the other peer has
-                        // already seen
-                        conn.reliable_queue.retain(|v| v.message_id > msgs.ack);
-
-                        // Process all messages that have not been previously seen
-                        // before, that is to say the connection's ack value is
-                        // lower than the message's id
-                        for reliable_msg in &msgs.reliable {
-                            if reliable_msg.message_id > conn.ack {
-                                messages.push(PeerMessage::Message(
-                                    conn.clone(),
-                                    reliable_msg.message.clone(),
-                                ));
-                            }
-                        }
-
-                        // Set connection's ack value to be at least the highest
-                        // message_id that was received, so that we inform the
-                        // remote peer not to send these messages again.
-                        conn.ack = msgs
-                            .reliable
-                            .iter()
-                            .map(|r| r.message_id)
-                            .max()
-                            .unwrap_or(0)
-                            .max(conn.ack);
-
-                        // Always process all unreliable messages
-                        for unreliable_msg in &msgs.unreliable {
-                            messages.push(PeerMessage::Message(
-                                conn.clone(),
-                                unreliable_msg.message.clone(),
-                            ));
-                        }
-                    }
-                }
-            }
         }
 
         messages
@@ -391,7 +247,7 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
     }
 
     /// Remove any connections that have closed or errored.
-    pub fn purge_old_connections(&mut self) -> Vec<PeerMessage<T>> {
+    pub fn purge_old_connections(&mut self) -> Vec<PeerMessage> {
         let mut messages = Vec::new();
 
         let now = Instant::now();
@@ -417,14 +273,6 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
 
         messages
     }
-
-    pub fn connections(&self) -> Vec<Connection<T>> {
-        self.connections.values().cloned().collect()
-    }
-
-    pub fn set_accepting_connections(&mut self, accepting: bool) {
-        self.accepting_connections = accepting;
-    }
 }
 
 /// Wrap UdpSocket with some helper methods
@@ -450,11 +298,7 @@ impl UdpWrapper {
     }
 
     /// Send a Package to a remote peer at addr
-    fn send_to<T: Serialize>(
+    fn send_to(&self, addr: SocketAddr, package: Package) -> Result<(), SendError> {
-        &self,
-        addr: SocketAddr,
-        package: Package<T>,
-    ) -> Result<usize, SendError> {
         if let Some(socket) = &self.socket {
             let mut buf = Vec::new();
             let cursor = Cursor::new(&mut buf);
@@ -466,7 +310,7 @@ impl UdpWrapper {
             socket
                 .send_to(&buf, addr)
                 .map_err(|e| SendError::SendError(e))
-                .map(|_| buf.len())
+                .map(|_| ())
         } else {
             Err(SendError::SocketDisconnected)
         }
@@ -474,7 +318,7 @@ impl UdpWrapper {
 }
 
 /// Represents a single connection from the peer to another peer
-pub struct Connection<T: Clone> {
+pub struct Connection {
     /// Address of the connection
     pub address: SocketAddr,
     /// "ping", as in the time between the last sent ping and the last received
@@ -486,24 +330,12 @@ pub struct Connection<T: Clone> {
     last_recv_close: Instant,
     closing_since: Instant,
 
-    message_counter: u64,
-    ack: u64,
-    reliable_queue: Vec<Message<T>>,
-    last_msg_send: Instant,
-
     /// What state is the connection currently in?
     pub state: ConnectionState,
     error: Option<ConnectionError>,
-
-    // Stats
-    messages_sent: usize,
-    messages_received: usize,
-    messages_expected: usize,
-    bytes_tx: usize,
-    bytes_rx: usize,
 }
 
-impl<T: Clone> Clone for Connection<T> {
+impl Clone for Connection {
     fn clone(&self) -> Self {
         Self {
             address: self.address.clone(),
@@ -512,27 +344,14 @@ impl<T: Clone> Clone for Connection<T> {
             last_recv_ping: self.last_recv_ping.clone(),
             last_recv_close: self.last_recv_close.clone(),
             closing_since: self.closing_since.clone(),
-
-            message_counter: self.message_counter,
-            ack: self.ack,
-            reliable_queue: self.reliable_queue.clone(),
-            last_msg_send: self.last_msg_send.clone(),
-
             state: self.state.clone(),
             error: None,
-
-            // Stats
-            messages_sent: self.messages_sent,
-            messages_received: self.messages_received,
-            messages_expected: self.messages_expected,
-            bytes_tx: self.bytes_tx,
-            bytes_rx: self.bytes_rx,
         }
     }
 }
 
-impl<T: Clone> Connection<T> {
+impl Connection {
-    pub fn from(address: SocketAddr, state: ConnectionState) -> Connection<T> {
+    pub fn from(address: SocketAddr, state: ConnectionState) -> Connection {
         Connection {
             address,
             ping: Duration::default(),
@@ -540,29 +359,8 @@ impl<T: Clone> Connection<T> {
             last_recv_ping: Instant::now(),
             last_recv_close: Instant::now(),
             closing_since: Instant::now(),
-
-            message_counter: 1,
-            ack: 0,
-            reliable_queue: Vec::new(),
-            last_msg_send: Instant::now() - Duration::from_hours(1),
-
             state,
             error: None,
-
-            messages_sent: 0,
-            messages_received: 0,
-            messages_expected: 0,
-            bytes_tx: 0,
-            bytes_rx: 0,
-        }
-    }
-
-    pub fn statistics(&self) -> NetStats {
-        NetStats {
-            bytes_tx: self.bytes_tx,
-            bytes_rx: self.bytes_rx,
-            messages_received: self.messages_received,
-            messages_expected: self.messages_expected,
         }
     }
 }

src/lib.rs

@@ -10,19 +10,16 @@ use std::{
     time::Duration,
 };
 
-use serde::{Serialize, de::DeserializeOwned};
 use thiserror::*;
 
 use crate::{
     connections::{Connection, ConnectionError, ConnectionManager},
     listener::{Listener, ListenerMessage},
-    stats::NetStats,
 };
 
 pub mod connections;
 pub(crate) mod listener;
 pub(crate) mod package;
-pub mod stats;
 
 /// Error for the Peer
 #[derive(Error, Debug)]
@@ -34,15 +31,13 @@ pub enum PeerError {
 }
 
 /// Possible messages from occurring events from the peer
-pub enum PeerMessage<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> {
+pub enum PeerMessage {
     /// A new connection has been connected
-    NewConnection(Connection<T>),
+    NewConnection(Connection),
     /// An existing connection has disconnected, with an optional error
-    Disconnected(Connection<T>, Option<ConnectionError>),
+    Disconnected(Connection, Option<ConnectionError>),
     /// The Peer has been closed
     Closed,
-    /// A single message of type T
-    Message(Connection<T>, T),
 }
 
 /// Optional configuration available for peers, mainly configuration of
@@ -52,7 +47,6 @@ pub struct PeerConfig {
     ping_interval: Duration,
     timeout: Duration,
     disconnect_timeout: Duration,
-    message_retry: Duration,
 }
 
 impl Default for PeerConfig {
@@ -61,7 +55,6 @@ impl Default for PeerConfig {
             ping_interval: Duration::from_millis(100),
             timeout: Duration::from_millis(2000),
             disconnect_timeout: Duration::from_millis(500),
-            message_retry: Duration::from_millis(100),
         }
     }
 }
@@ -90,29 +83,19 @@ impl PeerConfig {
             ..self
         }
     }
-
-    /// Sets the duration which is awaited before any queued reliable messages
-    /// are sent again. This timer is also reset every time any new messages are
-    /// sent manually.
-    pub fn with_retry(self, retry: Duration) -> PeerConfig {
-        PeerConfig {
-            message_retry: retry,
-            ..self
-        }
-    }
 }
 
-pub struct Peer<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static> {
+pub struct Peer {
-    connection_mgr: ConnectionManager<T>,
+    connection_mgr: ConnectionManager,
     closed: Arc<AtomicBool>,
-    receiver: Receiver<ListenerMessage<T>>,
+    receiver: Receiver<ListenerMessage>,
-    messages: VecDeque<PeerMessage<T>>,
+    messages: VecDeque<PeerMessage>,
 }
 
-impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static> Peer<T> {
+impl Peer {
     /// Bind given port to listen to incoming messages. Creates a new Peer that
     /// can establish new connections.
-    pub fn listen(port: Option<u16>, config: PeerConfig) -> Result<Peer<T>, PeerError> {
+    pub fn listen(port: Option<u16>, config: PeerConfig) -> Result<Peer, PeerError> {
         let socket = Arc::new(
             UdpSocket::bind(SocketAddr::from(([0, 0, 0, 0], port.unwrap_or(0))))
                 .map_err(|e| PeerError::BindError(e))?,
@@ -150,7 +133,7 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
     /// possible events, such as connections and messages. Should be called as
     /// often as possible.
     #[must_use]
-    pub fn poll(&mut self) -> Result<Option<PeerMessage<T>>, PeerError> {
+    pub fn poll(&mut self) -> Result<Option<PeerMessage>, PeerError> {
         if self.connection_mgr.udp.is_closed() {
             if self.closed.load(Ordering::Acquire) {
                 self.messages.push_back(PeerMessage::Closed);
@@ -159,7 +142,6 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
         }
 
         self.connection_mgr.send_pings();
-        self.connection_mgr.send_queued_messages();
 
         let mut msg;
         while {
@@ -171,12 +153,9 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
             msg.is_some()
         } {
             match msg.unwrap() {
-                ListenerMessage::Package(package, socket_addr, bytes) => {
+                ListenerMessage::Package(package, socket_addr) => {
-                    self.messages.extend(self.connection_mgr.handle_package(
+                    self.messages
-                        package,
+                        .extend(self.connection_mgr.handle_package(package, &socket_addr));
-                        &socket_addr,
-                        bytes,
-                    ));
                 }
                 ListenerMessage::PackageError(listener_error, socket_addr) => {
                     println!("Error: {}", listener_error);
@@ -201,57 +180,8 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
         Ok(self.messages.pop_front())
     }
 
-    /// Send a reliable message, which is retried until the remote peer
-    /// acknowledges it.
-    pub fn send_reliable(&mut self, to: &SocketAddr, message: T) {
-        self.connection_mgr.send(to, message, true);
-    }
-
-    /// Sends an unreliable message, which is never retried again.
-    pub fn send_unreliable(&mut self, to: &SocketAddr, message: T) {
-        self.connection_mgr.send(to, message, false);
-    }
-
-    /// Send a reliable message to every connection currently connected
-    pub fn broadcast_reliable(&mut self, message: T) {
-        for connection in self.connections() {
-            self.connection_mgr
-                .send(&connection.address, message.clone(), true);
-        }
-    }
-
-    /// Send an unreliable message to every connection currently connected
-    pub fn broadcast_unreliable(&mut self, message: T) {
-        for connection in self.connections() {
-            self.connection_mgr
-                .send(&connection.address, message.clone(), false);
-        }
-    }
-
     /// Close the Peer, disconnecting all connections.
     pub fn close(&mut self) {
         self.connection_mgr.close();
     }
-
-    /// List of all current connections
-    pub fn connections(&self) -> Vec<Connection<T>> {
-        self.connection_mgr.connections()
-    }
-
-    pub fn set_accepting_connections(&mut self, accepting: bool) {
-        self.connection_mgr.set_accepting_connections(accepting);
-    }
-
-    pub fn statistics(&self) -> NetStats {
-        let conns = self.connection_mgr.connections();
-        let all_stats = conns.iter().map(|c| c.statistics());
-        let mut total_stats = NetStats::default();
-        for stats in all_stats {
-            total_stats.bytes_rx += stats.bytes_rx;
-            total_stats.bytes_tx += stats.bytes_tx;
-            total_stats.messages_expected += stats.messages_expected;
-            total_stats.messages_received += stats.messages_received;
-        }
-        total_stats
-    }
 }

src/listener.rs

@@ -1,11 +1,9 @@
 use std::{
     io::Cursor,
-    marker::PhantomData,
     net::{SocketAddr, UdpSocket},
     sync::{Arc, mpsc::Sender},
 };
 
-use serde::{Serialize, de::DeserializeOwned};
 use thiserror::Error;
 
 use crate::package::Package;
@@ -14,19 +12,14 @@ pub const DATAGRAM_SIZE: usize = 102400;
 pub type Datagram = [u8; DATAGRAM_SIZE];
 
 /// Listener thread for Peer's UdpSocket
-pub struct Listener<T: Clone + Serialize + DeserializeOwned> {
+pub struct Listener {
     socket: Arc<UdpSocket>,
-    sender: Sender<ListenerMessage<T>>,
+    sender: Sender<ListenerMessage>,
-    pd: PhantomData<T>,
 }
 
-impl<'a, T: Clone + Serialize + DeserializeOwned> Listener<T> {
+impl Listener {
-    pub fn new(socket: Arc<UdpSocket>, sender: Sender<ListenerMessage<T>>) -> Listener<T> {
+    pub fn new(socket: Arc<UdpSocket>, sender: Sender<ListenerMessage>) -> Listener {
-        Listener {
+        Listener { socket, sender }
-            socket,
-            sender,
-            pd: PhantomData::default(),
-        }
     }
 
     /// Read from the socket, parse packages and emit messages.
@@ -36,12 +29,9 @@ impl<'a, T: Clone + Serialize + DeserializeOwned> Listener<T> {
             Ok((num_bytes, from_addr)) => {
                 let mut ciborium_buf: Datagram = [0; _];
                 let bytes = Cursor::new(&mut buffer[..num_bytes]);
-                let res =
+                let res = ciborium::from_reader_with_buffer::<Package, _>(bytes, &mut ciborium_buf);
-                    ciborium::from_reader_with_buffer::<Package<T>, _>(bytes, &mut ciborium_buf);
                 match res {
-                    Ok(pkg) => self
+                    Ok(pkg) => self.sender.send(ListenerMessage::Package(pkg, from_addr)),
-                        .sender
-                        .send(ListenerMessage::Package(pkg, from_addr, num_bytes)),
                     Err(_) => self.sender.send(ListenerMessage::PackageError(
                         ListenerError::ParseError,
                         from_addr,
@@ -62,8 +52,8 @@ pub enum ListenerError {
     ParseError,
 }
 
-pub enum ListenerMessage<T: Clone + Serialize + DeserializeOwned> {
+pub enum ListenerMessage {
-    Package(Package<T>, SocketAddr, usize),
+    Package(Package, SocketAddr),
     PackageError(ListenerError, SocketAddr),
     Error(std::io::Error),
 }

src/package.rs

@@ -1,24 +1,9 @@
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Serialize, Deserialize)]
-pub enum Package<T> {
+pub enum Package {
     Hello,
     Ping,
     Pong,
     Close,
-    Messages(Messages<T>),
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct Messages<T> {
-    pub ack: u64,
-    pub messages_sent: usize,
-    pub reliable: Vec<Message<T>>,
-    pub unreliable: Vec<Message<T>>,
-}
-
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct Message<T> {
-    pub message_id: u64,
-    pub message: T,
 }

src/stats.rs

@@ -1,7 +0,0 @@
-#[derive(Default)]
-pub struct NetStats {
-    pub bytes_tx: usize,
-    pub bytes_rx: usize,
-    pub messages_received: usize,
-    pub messages_expected: usize,
-}