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 bounded channels by default using the capacity set in `MqttOptions`,
+  with opt-in unbounded channels via `.unbounded()`.
+
 ### 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 backpressure
+    let (mut client, mut connection) = Client::builder(mqtt_opts).unbounded().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

@@ -2,6 +2,7 @@
 //! async eventloop.
 use std::time::Duration;
 
+use crate::eventloop::ChannelCapacity;
 use crate::mqttbytes::{v4::*, QoS};
 use crate::{valid_filter, valid_topic, ConnectionError, Event, EventLoop, MqttOptions, Request};
 
@@ -32,6 +33,62 @@ impl From<TrySendError<Request>> for ClientError {
     }
 }
 
+/// Builder for synchronous and asynchronous MQTT v4 clients.
+///
+/// The request channel is bounded by default, using the capacity configured
+/// in [`MqttOptions`]. Call [`capacity`](Self::capacity) to override it or
+/// [`unbounded`](Self::unbounded) to opt out of backpressure entirely.
+pub struct ClientBuilder {
+    options: MqttOptions,
+    capacity: ChannelCapacity,
+}
+
+impl ClientBuilder {
+    /// Create a new builder. The channel capacity defaults to the value set
+    /// in [`MqttOptions::request_channel_capacity`].
+    pub fn new(options: MqttOptions) -> Self {
+        let capacity = ChannelCapacity::Bounded(options.request_channel_capacity());
+        Self { options, capacity }
+    }
+
+    /// Set a bounded request channel with the given capacity.
+    /// A capacity of `0` creates a rendezvous channel.
+    pub fn capacity(mut self, cap: usize) -> Self {
+        self.capacity = ChannelCapacity::Bounded(cap);
+        self
+    }
+
+    /// Use an unbounded request channel.
+    /// This removes backpressure from the client. Prefer bounded channels
+    /// unless you have a specific reason to allow unbounded queuing.
+    pub fn unbounded(mut self) -> Self {
+        self.capacity = ChannelCapacity::Unbounded;
+        self
+    }
+
+    /// Build an [`AsyncClient`] and its [`EventLoop`].
+    pub fn build_async(self) -> (AsyncClient, EventLoop) {
+        let eventloop = EventLoop::new(self.options, self.capacity);
+        let request_tx = eventloop.requests_tx.clone();
+        let client = AsyncClient::from_senders(request_tx);
+        (client, eventloop)
+    }
+
+    /// Build a [`Client`] and its [`Connection`].
+    pub fn build(self) -> (Client, Connection) {
+        let (async_client, eventloop) = self.build_async();
+        let client = Client {
+            client: async_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`),
@@ -45,16 +102,15 @@ pub struct AsyncClient {
 }
 
 impl AsyncClient {
-    /// Create a new `AsyncClient`.
-    ///
-    /// `cap` specifies the capacity of the bounded async channel.
+    /// Deprecated. Use [`AsyncClient::builder`] instead.
+    #[deprecated(since = "0.26.0", note = "Use AsyncClient::builder 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)
+    /// Returns a [`ClientBuilder`] for constructing an MQTT v4 async client.
+    pub fn builder(options: MqttOptions) -> ClientBuilder {
+        ClientBuilder::new(options)
     }
 
     /// Create a new `AsyncClient` from a channel `Sender`.
@@ -253,19 +309,15 @@ pub struct Client {
 }
 
 impl Client {
-    /// Create a new `Client`
-    ///
-    /// `cap` specifies the capacity of the bounded async channel.
+    /// Deprecated. Use [`Client::builder`] instead.
+    #[deprecated(since = "0.26.0", note = "Use Client::builder 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)
+    /// Returns a [`ClientBuilder`] for constructing an MQTT v4 sync client.
+    pub fn builder(options: MqttOptions) -> ClientBuilder {
+        ClientBuilder::new(options)
     }
 
     /// Create a new `Client` from a channel `Sender`.
@@ -532,7 +584,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};
@@ -31,6 +31,19 @@ use {
 #[cfg(feature = "proxy")]
 use crate::proxy::ProxyError;
 
+/// Controls the capacity of the client request channel.
+///
+/// Defaults to [`Bounded`] which applies backpressure to the client.
+/// Use [`Unbounded`] only when you have a specific reason to remove backpressure.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ChannelCapacity {
+    /// Bounded channel with the given capacity.
+    /// A capacity of `0` creates a rendezvous channel.
+    Bounded(usize),
+    /// Unbounded channel with no backpressure.
+    Unbounded,
+}
+
 /// Critical errors during eventloop polling
 #[derive(Debug, thiserror::Error)]
 pub enum ConnectionError {
@@ -99,8 +112,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: ChannelCapacity) -> EventLoop {
+        let (requests_tx, requests_rx) = match cap {
+            ChannelCapacity::Bounded(n) => bounded(n),
+            ChannelCapacity::Unbounded => unbounded(),
+        };
         let pending = VecDeque::new();
         let max_inflight = mqtt_options.inflight;
         let manual_acks = mqtt_options.manual_acks;

rumqttc/src/v5/client.rs

@@ -2,6 +2,7 @@
 //! async eventloop.
 use std::time::Duration;
 
+use super::eventloop::ChannelCapacity;
 use super::mqttbytes::v5::{
     Filter, PubAck, PubRec, Publish, PublishProperties, Subscribe, SubscribeProperties,
     Unsubscribe, UnsubscribeProperties,
@@ -37,6 +38,62 @@ impl From<TrySendError<Request>> for ClientError {
     }
 }
 
+/// Builder for synchronous and asynchronous MQTT v5 clients.
+///
+/// The request channel is bounded by default, using the capacity configured
+/// in [`MqttOptions`]. Call [`capacity`](Self::capacity) to override it or
+/// [`unbounded`](Self::unbounded) to opt out of backpressure entirely.
+pub struct AsyncClientBuilder {
+    options: MqttOptions,
+    capacity: ChannelCapacity,
+}
+
+impl AsyncClientBuilder {
+    /// Create a new builder. The channel capacity defaults to the value set
+    /// in [`MqttOptions::request_channel_capacity`].
+    pub fn new(options: MqttOptions) -> Self {
+        let capacity = ChannelCapacity::Bounded(options.request_channel_capacity());
+        Self { options, capacity }
+    }
+
+    /// Set a bounded request channel with the given capacity.
+    /// A capacity of `0` creates a rendezvous channel.
+    pub fn capacity(mut self, cap: usize) -> Self {
+        self.capacity = ChannelCapacity::Bounded(cap);
+        self
+    }
+
+    /// Use an unbounded request channel.
+    /// This removes backpressure from the client. Prefer bounded channels
+    /// unless you have a specific reason to allow unbounded queuing.
+    pub fn unbounded(mut self) -> Self {
+        self.capacity = ChannelCapacity::Unbounded;
+        self
+    }
+
+    /// Build an [`AsyncClient`] and its [`EventLoop`].
+    pub fn build_async(self) -> (AsyncClient, EventLoop) {
+        let eventloop = EventLoop::new(self.options, self.capacity);
+        let request_tx = eventloop.requests_tx.clone();
+        let client = AsyncClient::from_senders(request_tx);
+        (client, eventloop)
+    }
+
+    /// Build a [`Client`] and its [`Connection`].
+    pub fn build(self) -> (Client, Connection) {
+        let (async_client, eventloop) = self.build_async();
+        let client = Client {
+            client: async_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`),
@@ -50,16 +107,15 @@ pub struct AsyncClient {
 }
 
 impl AsyncClient {
-    /// Create a new `AsyncClient`.
-    ///
-    /// `cap` specifies the capacity of the bounded async channel.
+    /// Deprecated. Use [`AsyncClient::builder`] instead.
+    #[deprecated(since = "0.26.0", note = "Use AsyncClient::builder 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)
+    /// Returns an [`AsyncClientBuilder`] for constructing an MQTT v5 async client.
+    pub fn builder(options: MqttOptions) -> AsyncClientBuilder {
+        AsyncClientBuilder::new(options)
     }
 
     /// Create a new `AsyncClient` from a channel `Sender`.
@@ -469,20 +525,15 @@ pub struct Client {
 }
 
 impl Client {
-    /// Create a new `Client`
-    ///
-    /// `cap` specifies the capacity of the bounded async channel.
+    /// Deprecated. Use [`Client::builder`] instead.
+    #[deprecated(since = "0.26.0", note = "Use Client::builder 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)
+    /// Returns an [`AsyncClientBuilder`] for constructing an MQTT v5 sync client.
+    pub fn builder(options: MqttOptions) -> AsyncClientBuilder {
+        AsyncClientBuilder::new(options)
     }
 
     /// Create a new `Client` from a channel `Sender`.
@@ -882,7 +933,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};
 
@@ -31,6 +31,19 @@ use {
 #[cfg(feature = "proxy")]
 use crate::proxy::ProxyError;
 
+/// Controls the capacity of the client request channel.
+///
+/// Defaults to [`Bounded`] which applies backpressure to the client.
+/// Use [`Unbounded`] only when you have a specific reason to remove backpressure.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ChannelCapacity {
+    /// Bounded channel with the given capacity.
+    /// A capacity of `0` creates a rendezvous channel.
+    Bounded(usize),
+    /// Unbounded channel with no backpressure.
+    Unbounded,
+}
+
 /// Critical errors during eventloop polling
 #[derive(Debug, thiserror::Error)]
 pub enum ConnectionError {
@@ -96,8 +109,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: ChannelCapacity) -> EventLoop {
+        let (requests_tx, requests_rx) = match cap {
+            ChannelCapacity::Bounded(n) => bounded(n),
+            ChannelCapacity::Unbounded => unbounded(),
+        };
         let pending = VecDeque::new();
         let inflight_limit = options.outgoing_inflight_upper_limit.unwrap_or(u16::MAX);
         let manual_acks = options.manual_acks;