feat(rumqttc): add support for unbounded client

Signed-off-by: Jyotiraditya Panda <jyotiraditya@aospa.co>

Author: imjyotiraditya

rumqttc/CHANGELOG.md

@@ -8,8 +8,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- `ClientBuilder` and `AsyncClientBuilder` for both v4 and v5 clients,
+  enabling unbounded channels by default with opt-in bounded capacity
+  via `.capacity(n)`.
+
 ### Changed
 ### Deprecated
+- `Client::new(options, cap)` and `AsyncClient::new(options, cap)` —
+  use `Client::builder(options).build()` or
+  `Client::builder(options).capacity(cap).build()` instead.
+
 ### Removed
 ### Fixed 
 ### Security

rumqttc/examples/unbounded.rs

@@ -0,0 +1,25 @@
+use rumqttc::{Client, MqttOptions, QoS};
+use std::thread;
+use std::time::Duration;
+
+fn main() {
+    let mqtt_opts = MqttOptions::new("test-1", "localhost", 1883);
+
+    // unbounded channel — no capacity limit
+    let (mut client, mut connection) = Client::builder(mqtt_opts).build();
+
+    client.subscribe("hello/rumqtt", QoS::AtMostOnce).unwrap();
+
+    thread::spawn(move || {
+        for i in 0..10 {
+            client
+                .publish("hello/rumqtt", QoS::AtLeastOnce, false, vec![1; i])
+                .unwrap();
+            thread::sleep(Duration::from_millis(100));
+        }
+    });
+
+    for notification in connection.iter() {
+        println!("Notification = {:?}", notification);
+    }
+}

rumqttc/src/client.rs

@@ -32,6 +32,42 @@ impl From<TrySendError<Request>> for ClientError {
     }
 }
 
+pub struct ClientBuilder {
+    options: MqttOptions,
+    capacity: Option<usize>,
+}
+
+impl ClientBuilder {
+    pub fn new(options: MqttOptions) -> Self {
+        Self {
+            options,
+            capacity: None,
+        }
+    }
+
+    pub fn capacity(mut self, cap: usize) -> Self {
+        self.capacity = Some(cap);
+        self
+    }
+
+    pub fn build_async(self) -> (AsyncClient, EventLoop) {
+        let eventloop = EventLoop::new(self.options, self.capacity);
+        let request_tx = eventloop.requests_tx.clone();
+        (AsyncClient { request_tx }, eventloop)
+    }
+
+    pub fn build(self) -> (Client, Connection) {
+        let (client, eventloop) = self.build_async();
+        let client = Client { client };
+        let runtime = runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let connection = Connection::new(eventloop, runtime);
+        (client, connection)
+    }
+}
+
 /// An asynchronous client, communicates with MQTT `EventLoop`.
 ///
 /// This is cloneable and can be used to asynchronously [`publish`](`AsyncClient::publish`),
@@ -47,14 +83,17 @@ pub struct AsyncClient {
 impl AsyncClient {
     /// Create a new `AsyncClient`.
     ///
-    /// `cap` specifies the capacity of the bounded async channel.
+    /// # Deprecation
+    /// Use [`ClientBuilder`] instead for bounded/unbounded control.
+    /// `Client::builder(options).capacity(cap).build_async()` for bounded,
+    /// `Client::builder(options).build_async()` for unbounded.
+    #[deprecated(since = "0.26.0", note = "Use ClientBuilder instead")]
     pub fn new(options: MqttOptions, cap: usize) -> (AsyncClient, EventLoop) {
-        let eventloop = EventLoop::new(options, cap);
-        let request_tx = eventloop.requests_tx.clone();
-
-        let client = AsyncClient { request_tx };
+        ClientBuilder::new(options).capacity(cap).build_async()
+    }
 
-        (client, eventloop)
+    pub fn builder(options: MqttOptions) -> ClientBuilder {
+        ClientBuilder::new(options)
     }
 
     /// Create a new `AsyncClient` from a channel `Sender`.
@@ -253,19 +292,17 @@ pub struct Client {
 }
 
 impl Client {
-    /// Create a new `Client`
+    /// Create a new `Client`.
     ///
-    /// `cap` specifies the capacity of the bounded async channel.
+    /// # Deprecation
+    /// Use [`ClientBuilder`] instead.
+    #[deprecated(since = "0.26.0", note = "Use ClientBuilder instead")]
     pub fn new(options: MqttOptions, cap: usize) -> (Client, Connection) {
-        let (client, eventloop) = AsyncClient::new(options, cap);
-        let client = Client { client };
-        let runtime = runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .unwrap();
+        ClientBuilder::new(options).capacity(cap).build()
+    }
 
-        let connection = Connection::new(eventloop, runtime);
-        (client, connection)
+    pub fn builder(options: MqttOptions) -> ClientBuilder {
+        ClientBuilder::new(options)
     }
 
     /// Create a new `Client` from a channel `Sender`.
@@ -532,7 +569,7 @@ mod test {
             .set_keep_alive(Duration::from_secs(5))
             .set_last_will(will);
 
-        let (_, mut connection) = Client::new(mqttoptions, 10);
+        let (_, mut connection) = Client::builder(mqttoptions).capacity(10).build();
         let _ = connection.iter();
         let _ = connection.iter();
     }

rumqttc/src/eventloop.rs

@@ -4,7 +4,7 @@ use crate::{MqttOptions, Outgoing};
 
 use crate::framed::AsyncReadWrite;
 use crate::mqttbytes::v4::*;
-use flume::{bounded, Receiver, Sender};
+use flume::{bounded, unbounded, Receiver, Sender};
 use tokio::net::{lookup_host, TcpSocket, TcpStream};
 use tokio::select;
 use tokio::time::{self, Instant, Sleep};
@@ -99,8 +99,11 @@ impl EventLoop {
     ///
     /// When connection encounters critical errors (like auth failure), user has a choice to
     /// access and update `options`, `state` and `requests`.
-    pub fn new(mqtt_options: MqttOptions, cap: usize) -> EventLoop {
-        let (requests_tx, requests_rx) = bounded(cap);
+    pub fn new(mqtt_options: MqttOptions, cap: Option<usize>) -> EventLoop {
+        let (requests_tx, requests_rx) = match cap {
+            Some(cap) => bounded(cap),
+            None => unbounded(),
+        };
         let pending = VecDeque::new();
         let max_inflight = mqtt_options.inflight;
         let manual_acks = mqtt_options.manual_acks;

rumqttc/src/v5/client.rs

@@ -37,6 +37,41 @@ impl From<TrySendError<Request>> for ClientError {
     }
 }
 
+pub struct AsyncClientBuilder {
+    options: MqttOptions,
+    capacity: Option<usize>,
+}
+
+impl AsyncClientBuilder {
+    pub fn new(options: MqttOptions) -> Self {
+        Self {
+            options,
+            capacity: None,
+        }
+    }
+
+    pub fn capacity(mut self, cap: usize) -> Self {
+        self.capacity = Some(cap);
+        self
+    }
+
+    pub fn build_async(self) -> (AsyncClient, EventLoop) {
+        let eventloop = EventLoop::new(self.options, self.capacity);
+        let request_tx = eventloop.requests_tx.clone();
+        (AsyncClient { request_tx }, eventloop)
+    }
+
+    pub fn build(self) -> (Client, Connection) {
+        let (client, eventloop) = self.build_async();
+        let client = Client { client };
+        let runtime = runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        (client, Connection::new(eventloop, runtime))
+    }
+}
+
 /// An asynchronous client, communicates with MQTT `EventLoop`.
 ///
 /// This is cloneable and can be used to asynchronously [`publish`](`AsyncClient::publish`),
@@ -52,14 +87,17 @@ pub struct AsyncClient {
 impl AsyncClient {
     /// Create a new `AsyncClient`.
     ///
-    /// `cap` specifies the capacity of the bounded async channel.
+    /// # Deprecation
+    /// Use [`AsyncClientBuilder`] instead for bounded/unbounded control.
+    /// `Client::builder(options).capacity(cap).build_async()` for bounded,
+    /// `Client::builder(options).build_async()` for unbounded.
+    #[deprecated(since = "0.26.0", note = "Use AsyncClientBuilder instead")]
     pub fn new(options: MqttOptions, cap: usize) -> (AsyncClient, EventLoop) {
-        let eventloop = EventLoop::new(options, cap);
-        let request_tx = eventloop.requests_tx.clone();
-
-        let client = AsyncClient { request_tx };
+        AsyncClientBuilder::new(options).capacity(cap).build_async()
+    }
 
-        (client, eventloop)
+    pub fn builder(options: MqttOptions) -> AsyncClientBuilder {
+        AsyncClientBuilder::new(options)
     }
 
     /// Create a new `AsyncClient` from a channel `Sender`.
@@ -469,20 +507,17 @@ pub struct Client {
 }
 
 impl Client {
-    /// Create a new `Client`
+    /// Create a new `Client`.
     ///
-    /// `cap` specifies the capacity of the bounded async channel.
+    /// # Deprecation
+    /// Use [`AsyncClientBuilder`] instead.
+    #[deprecated(since = "0.26.0", note = "Use AsyncClientBuilder instead")]
     pub fn new(options: MqttOptions, cap: usize) -> (Client, Connection) {
-        let (client, eventloop) = AsyncClient::new(options, cap);
-        let client = Client { client };
-
-        let runtime = runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .unwrap();
+        AsyncClientBuilder::new(options).capacity(cap).build()
+    }
 
-        let connection = Connection::new(eventloop, runtime);
-        (client, connection)
+    pub fn builder(options: MqttOptions) -> AsyncClientBuilder {
+        AsyncClientBuilder::new(options)
     }
 
     /// Create a new `Client` from a channel `Sender`.
@@ -882,7 +917,7 @@ mod test {
             .set_keep_alive(Duration::from_secs(5))
             .set_last_will(will);
 
-        let (_, mut connection) = Client::new(mqttoptions, 10);
+        let (_, mut connection) = Client::builder(mqttoptions).capacity(10).build();
         let _ = connection.iter();
         let _ = connection.iter();
     }

rumqttc/src/v5/eventloop.rs

@@ -4,7 +4,7 @@ use super::{Incoming, MqttOptions, MqttState, Outgoing, Request, StateError, Tra
 use crate::eventloop::socket_connect;
 use crate::framed::AsyncReadWrite;
 
-use flume::{bounded, Receiver, Sender};
+use flume::{bounded, unbounded, Receiver, Sender};
 use tokio::select;
 use tokio::time::{self, error::Elapsed, Instant, Sleep};
 
@@ -96,8 +96,11 @@ impl EventLoop {
     ///
     /// When connection encounters critical errors (like auth failure), user has a choice to
     /// access and update `options`, `state` and `requests`.
-    pub fn new(options: MqttOptions, cap: usize) -> EventLoop {
-        let (requests_tx, requests_rx) = bounded(cap);
+    pub fn new(options: MqttOptions, cap: Option<usize>) -> EventLoop {
+        let (requests_tx, requests_rx) = match cap {
+            Some(cap) => bounded(cap),
+            None => unbounded(),
+        };
         let pending = VecDeque::new();
         let inflight_limit = options.outgoing_inflight_upper_limit.unwrap_or(u16::MAX);
         let manual_acks = options.manual_acks;