chore(socket): consistent builder pattern across repo (#123)

Author: thedevbirb

Cargo.toml

@@ -108,9 +108,10 @@ all = "warn"
 [workspace.lints.rust]
 missing_debug_implementations = "warn"
 missing_docs = "warn"
-rust-2018-idioms = { level = "deny", priority = -1 }
-unreachable-pub = "warn"
-unused-must-use = "deny"
+rust_2018_idioms = { level = "deny", priority = -1 }
+unreachable_pub = "warn"
+unused_must_use = "deny"
+rust_2024_incompatible_pat = "warn"
 
 [workspace.lints.clippy]
 # These are some of clippy's nursery (i.e., experimental) lints that we like.

book/src/usage/authentication.md

@@ -55,7 +55,7 @@ async fn main() {
     // The identifier will be sent to the server when the connection is established.
     let mut req = ReqSocket::with_options(
         Tcp::default(),
-        ReqOptions::default().auth_token(Bytes::from("client1")),
+        ReqOptions::default().with_auth_token(Bytes::from("client1")),
     );
 
     ...

msg-socket/src/connection/backoff.rs

@@ -1,12 +1,16 @@
 use futures::{FutureExt, Stream};
-use std::{pin::Pin, task::Poll, time::Duration};
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+    time::Duration,
+};
 use tokio::time::sleep;
 
 /// Helper trait alias for backoff streams.
 /// We define any stream that yields `Duration`s as a backoff
 pub trait Backoff: Stream<Item = Duration> + Unpin {}
 
-/// Blanket implementation of `Backoff` for any stream that yields `Duration`s.
+// Blanket implementation of `Backoff` for any stream that yields `Duration`s.
 impl<T> Backoff for T where T: Stream<Item = Duration> + Unpin {}
 
 /// A stream that yields exponentially increasing backoff durations.
@@ -23,6 +27,7 @@ pub struct ExponentialBackoff {
 }
 
 impl ExponentialBackoff {
+    /// Creates a new exponential backoff stream with the given initial duration and max retries.
     pub fn new(initial: Duration, max_retries: usize) -> Self {
         Self { retry_count: 0, max_retries, backoff: initial, timeout: None }
     }
@@ -38,38 +43,36 @@ impl Stream for ExponentialBackoff {
 
     /// Polls the exponential backoff stream. Returns `Poll::Ready` with the current backoff
     /// duration if the backoff timeout has elapsed, otherwise returns `Poll::Pending`.
-    fn poll_next(
-        self: Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.get_mut();
 
         loop {
-            if let Some(ref mut timeout) = this.timeout {
-                if timeout.poll_unpin(cx).is_ready() {
-                    // Timeout has elapsed, so reset the timeout and double the backoff
-                    this.backoff *= 2;
-                    this.retry_count += 1;
+            let Some(ref mut timeout) = this.timeout else {
+                // Set the initial timeout
+                this.reset_timeout();
+                continue;
+            };
+
+            if timeout.poll_unpin(cx).is_ready() {
+                // Timeout has elapsed, so reset the timeout and double the backoff
+                this.backoff *= 2;
+                this.retry_count += 1;
 
-                    // Close the stream
-                    if this.retry_count >= this.max_retries {
-                        return Poll::Ready(None);
-                    }
+                // Close the stream
+                if this.retry_count >= this.max_retries {
+                    return Poll::Ready(None);
+                }
 
-                    this.reset_timeout();
+                this.reset_timeout();
 
-                    // Wake up the task to poll the timeout again
-                    cx.waker().wake_by_ref();
+                // Wake up the task to poll the timeout again
+                cx.waker().wake_by_ref();
 
-                    // Return the current backoff duration
-                    return Poll::Ready(Some(this.backoff));
-                } else {
-                    // Timeout has not elapsed, so return pending
-                    return Poll::Pending;
-                }
+                // Return the current backoff duration
+                return Poll::Ready(Some(this.backoff));
             } else {
-                // Set initial timeout
-                this.reset_timeout();
+                // Timeout has not elapsed, so return pending
+                return Poll::Pending;
             }
         }
     }

msg-socket/src/connection/state.rs

@@ -1,12 +1,9 @@
-use msg_transport::Address;
-
-use super::Backoff;
-
 /// Abstraction to represent the state of a connection.
 ///
 /// * `C` is the channel type, which is used to send and receive generic messages.
 /// * `B` is the backoff type, used to control the backoff state for inactive connections.
-pub enum ConnectionState<C, B, A: Address> {
+/// * `A` is the address type, used to represent the address of the connection.
+pub enum ConnectionState<C, B, A> {
     Active {
         /// Channel to control the underlying connection. This is used to send
         /// and receive any kind of message in any direction.
@@ -19,15 +16,13 @@ pub enum ConnectionState<C, B, A: Address> {
     },
 }
 
-impl<C, B: Backoff, A: Address> ConnectionState<C, B, A> {
+impl<C, B, A> ConnectionState<C, B, A> {
     /// Returns `true` if the connection is active.
-    #[allow(unused)]
     pub fn is_active(&self) -> bool {
         matches!(self, Self::Active { .. })
     }
 
     /// Returns `true` if the connection is inactive.
-    #[allow(unused)]
     pub fn is_inactive(&self) -> bool {
         matches!(self, Self::Inactive { .. })
     }

msg-socket/src/pub/mod.rs

@@ -73,35 +73,35 @@ impl Default for PubOptions {
 
 impl PubOptions {
     /// Sets the maximum number of concurrent clients.
-    pub fn max_clients(mut self, max_clients: usize) -> Self {
+    pub fn with_max_clients(mut self, max_clients: usize) -> Self {
         self.max_clients = Some(max_clients);
         self
     }
 
     /// Sets the session channel buffer size. This is the amount of messages that can be buffered
     /// per session before messages start being dropped.
-    pub fn session_buffer_size(mut self, session_buffer_size: usize) -> Self {
+    pub fn with_session_buffer_size(mut self, session_buffer_size: usize) -> Self {
         self.session_buffer_size = session_buffer_size;
         self
     }
 
     /// Sets the maximum number of bytes that can be buffered in the session before being flushed.
     /// This internally sets [`Framed::set_backpressure_boundary`](tokio_util::codec::Framed).
-    pub fn backpressure_boundary(mut self, backpressure_boundary: usize) -> Self {
+    pub fn with_backpressure_boundary(mut self, backpressure_boundary: usize) -> Self {
         self.backpressure_boundary = backpressure_boundary;
         self
     }
 
     /// Sets the interval at which each session should be flushed. If this is `None`,
     /// the session will be flushed on every publish, which can add a lot of overhead.
-    pub fn flush_interval(mut self, flush_interval: std::time::Duration) -> Self {
+    pub fn with_flush_interval(mut self, flush_interval: std::time::Duration) -> Self {
         self.flush_interval = Some(flush_interval);
         self
     }
 
     /// Sets the minimum payload size in bytes for compression to be used. If the payload is smaller
     /// than this threshold, it will not be compressed.
-    pub fn min_compress_size(mut self, min_compress_size: usize) -> Self {
+    pub fn with_min_compress_size(mut self, min_compress_size: usize) -> Self {
         self.min_compress_size = min_compress_size;
         self
     }
@@ -224,7 +224,7 @@ mod tests {
 
         let mut sub_socket = SubSocket::with_options(
             Tcp::default(),
-            SubOptions::default().auth_token(Bytes::from("client1")),
+            SubOptions::default().with_auth_token(Bytes::from("client1")),
         );
 
         pub_socket.bind("0.0.0.0:0").await.unwrap();
@@ -250,7 +250,7 @@ mod tests {
 
         let mut sub_socket = SubSocket::with_options(
             Quic::default(),
-            SubOptions::default().auth_token(Bytes::from("client1")),
+            SubOptions::default().with_auth_token(Bytes::from("client1")),
         );
 
         pub_socket.bind("0.0.0.0:0").await.unwrap();
@@ -389,7 +389,7 @@ mod tests {
         let _ = tracing_subscriber::fmt::try_init();
 
         let mut pub_socket =
-            PubSocket::with_options(Tcp::default(), PubOptions::default().max_clients(1));
+            PubSocket::with_options(Tcp::default(), PubOptions::default().with_max_clients(1));
 
         pub_socket.bind("0.0.0.0:0").await.unwrap();
 

msg-socket/src/rep/mod.rs

@@ -50,19 +50,19 @@ impl Default for RepOptions {
 
 impl RepOptions {
     /// Sets the number of maximum concurrent clients.
-    pub fn max_clients(mut self, max_clients: usize) -> Self {
+    pub fn with_max_clients(mut self, max_clients: usize) -> Self {
         self.max_clients = Some(max_clients);
         self
     }
 
     /// Sets the minimum payload size for compression.
     /// If the payload is smaller than this value, it will not be compressed.
-    pub fn min_compress_size(mut self, min_compress_size: usize) -> Self {
+    pub fn with_min_compress_size(mut self, min_compress_size: usize) -> Self {
         self.min_compress_size = min_compress_size;
         self
     }
 
-    pub fn backpressure_boundary(mut self, backpressure_boundary: usize) -> Self {
+    pub fn with_backpressure_boundary(mut self, backpressure_boundary: usize) -> Self {
         self.backpressure_boundary = backpressure_boundary;
         self
     }
@@ -210,7 +210,7 @@ mod tests {
         // Initialize socket with a client ID. This will implicitly enable authentication.
         let mut req = ReqSocket::with_options(
             Tcp::default(),
-            ReqOptions::default().auth_token(Bytes::from("REQ")),
+            ReqOptions::default().with_auth_token(Bytes::from("REQ")),
         );
 
         req.connect(rep.local_addr().unwrap()).await.unwrap();
@@ -247,7 +247,8 @@ mod tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     async fn rep_max_connections() {
         let _ = tracing_subscriber::fmt::try_init();
-        let mut rep = RepSocket::with_options(Tcp::default(), RepOptions::default().max_clients(1));
+        let mut rep =
+            RepSocket::with_options(Tcp::default(), RepOptions::default().with_max_clients(1));
         rep.bind("127.0.0.1:0").await.unwrap();
         let addr = rep.local_addr().unwrap();
 
@@ -264,15 +265,19 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     async fn test_basic_reqrep_with_compression() {
-        let mut rep =
-            RepSocket::with_options(Tcp::default(), RepOptions::default().min_compress_size(0))
-                .with_compressor(SnappyCompressor);
+        let mut rep = RepSocket::with_options(
+            Tcp::default(),
+            RepOptions::default().with_min_compress_size(0),
+        )
+        .with_compressor(SnappyCompressor);
 
         rep.bind("0.0.0.0:4445").await.unwrap();
 
-        let mut req =
-            ReqSocket::with_options(Tcp::default(), ReqOptions::default().min_compress_size(0))
-                .with_compressor(GzipCompressor::new(6));
+        let mut req = ReqSocket::with_options(
+            Tcp::default(),
+            ReqOptions::default().with_min_compress_size(0),
+        )
+        .with_compressor(GzipCompressor::new(6));
 
         req.connect("0.0.0.0:4445").await.unwrap();
 

msg-socket/src/req/mod.rs

@@ -72,25 +72,25 @@ pub struct ReqOptions {
 
 impl ReqOptions {
     /// Sets the authentication token for the socket.
-    pub fn auth_token(mut self, auth_token: Bytes) -> Self {
+    pub fn with_auth_token(mut self, auth_token: Bytes) -> Self {
         self.auth_token = Some(auth_token);
         self
     }
 
     /// Sets the timeout for the socket.
-    pub fn timeout(mut self, timeout: Duration) -> Self {
+    pub fn with_timeout(mut self, timeout: Duration) -> Self {
         self.timeout = timeout;
         self
     }
 
     /// Enables blocking initial connections to the target.
-    pub fn blocking_connect(mut self) -> Self {
+    pub fn with_blocking_connect(mut self) -> Self {
         self.blocking_connect = true;
         self
     }
 
     /// Sets the backoff duration for the socket.
-    pub fn backoff_duration(mut self, backoff_duration: Duration) -> Self {
+    pub fn with_backoff_duration(mut self, backoff_duration: Duration) -> Self {
         self.backoff_duration = backoff_duration;
         self
     }
@@ -99,29 +99,29 @@ impl ReqOptions {
     /// throughput, but at the cost of higher latency. Note that this behaviour can be
     /// completely useless if the `backpressure_boundary` is set too low (which will trigger a
     /// flush before the interval is reached).
-    pub fn flush_interval(mut self, flush_interval: Duration) -> Self {
+    pub fn with_flush_interval(mut self, flush_interval: Duration) -> Self {
         self.flush_interval = Some(flush_interval);
         self
     }
 
     /// Sets the backpressure boundary for the socket. This is the maximum number of bytes that can
     /// be buffered in the session before being flushed. This internally sets
     /// [`Framed::set_backpressure_boundary`](tokio_util::codec::Framed).
-    pub fn backpressure_boundary(mut self, backpressure_boundary: usize) -> Self {
+    pub fn with_backpressure_boundary(mut self, backpressure_boundary: usize) -> Self {
         self.backpressure_boundary = backpressure_boundary;
         self
     }
 
     /// Sets the maximum number of retry attempts. If `None`, all connections will be retried
     /// indefinitely.
-    pub fn retry_attempts(mut self, retry_attempts: usize) -> Self {
+    pub fn with_retry_attempts(mut self, retry_attempts: usize) -> Self {
         self.retry_attempts = Some(retry_attempts);
         self
     }
 
     /// Sets the minimum payload size in bytes for compression to be used. If the payload is smaller
     /// than this threshold, it will not be compressed.
-    pub fn min_compress_size(mut self, min_compress_size: usize) -> Self {
+    pub fn with_min_compress_size(mut self, min_compress_size: usize) -> Self {
         self.min_compress_size = min_compress_size;
         self
     }

msg-socket/src/req/socket.rs

@@ -113,7 +113,6 @@ where
     /// Tries to connect to the target endpoint with the default options.
     /// A ReqSocket can only be connected to a single address.
     pub async fn try_connect(&mut self, endpoint: A) -> Result<(), ReqError> {
-        // Initialize communication channels
         let (to_driver, from_socket) = mpsc::channel(DEFAULT_BUFFER_SIZE);
 
         // TODO: Don't panic, return error

msg-socket/src/sub/mod.rs

@@ -71,27 +71,27 @@ pub struct SubOptions {
 impl SubOptions {
     /// Sets the authentication token for this socket. This will activate the authentication layer
     /// and send the token to the publisher.
-    pub fn auth_token(mut self, auth_token: Bytes) -> Self {
+    pub fn with_auth_token(mut self, auth_token: Bytes) -> Self {
         self.auth_token = Some(auth_token);
         self
     }
 
     /// Sets the ingress buffer size. This is the maximum amount of incoming messages that will be
     /// buffered. If the consumer cannot keep up with the incoming messages, messages will start
     /// being dropped.
-    pub fn ingress_buffer_size(mut self, ingress_buffer_size: usize) -> Self {
+    pub fn with_ingress_buffer_size(mut self, ingress_buffer_size: usize) -> Self {
         self.ingress_buffer_size = ingress_buffer_size;
         self
     }
 
     /// Sets the read buffer size. This sets the size of the read buffer for each session.
-    pub fn read_buffer_size(mut self, read_buffer_size: usize) -> Self {
+    pub fn with_read_buffer_size(mut self, read_buffer_size: usize) -> Self {
         self.read_buffer_size = read_buffer_size;
         self
     }
 
     /// Set the initial backoff for reconnecting to a publisher.
-    pub fn initial_backoff(mut self, initial_backoff: Duration) -> Self {
+    pub fn with_initial_backoff(mut self, initial_backoff: Duration) -> Self {
         self.initial_backoff = initial_backoff;
         self
     }