Option to configure TCP no_delay

rumqttc/CHANGELOG.md

@@ -14,12 +14,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * `ConnectionAborted` variant on `StateError` type to denote abrupt end to a connection
 * `set_session_expiry_interval` and `session_expiry_interval` methods on `MqttOptions`.
 * `Auth` packet as per MQTT5 standards
+* `tcp_nodelay` field on `NetworkOptions` type to enable configuring `tcp_nodelay` for the client
+* `set_tcp_nodelay` method on `Networkoptions`
 
 ### Changed
 
 * rename `N` as `AsyncReadWrite` to describe usage.
 * use `Framed` to encode/decode MQTT packets.
 * use `Login` to store credentials
+* check `tcp_nodelay` field of `NetworkOptions` when setting up the connection socket to set tcp_nodelay if the flag is set
 
 ### Deprecated
 

rumqttc/examples/asyncpubsub_nodelay.rs

@@ -0,0 +1,57 @@
+//! Same as asyncpubsub.rs but extended by configuring the client to not batch the messages
+//! sent to the broker over TCP
+
+use tokio::{task, time};
+
+use rumqttc::{AsyncClient, MqttOptions, QoS};
+use std::error::Error;
+use std::time::Duration;
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn Error>> {
+    pretty_env_logger::init();
+    // color_backtrace::install();
+
+    let mut mqttoptions = MqttOptions::new("test-1", "localhost", 1883);
+    mqttoptions.set_keep_alive(Duration::from_secs(5));
+
+    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+    // Configuring the client to not batch the messages sent to the broker over TCP
+    eventloop.network_options.set_tcp_nodelay(true);
+
+    task::spawn(async move {
+        requests(client).await;
+        time::sleep(Duration::from_secs(3)).await;
+    });
+
+    loop {
+        let event = eventloop.poll().await;
+        match &event {
+            Ok(v) => {
+                println!("Event = {v:?}");
+            }
+            Err(e) => {
+                println!("Error = {e:?}");
+                return Ok(());
+            }
+        }
+    }
+}
+
+async fn requests(client: AsyncClient) {
+    client
+        .subscribe("hello/world", QoS::AtMostOnce)
+        .await
+        .unwrap();
+
+    for i in 1..=10 {
+        client
+            .publish("hello/world", QoS::ExactlyOnce, false, vec![1; i])
+            .await
+            .unwrap();
+
+        time::sleep(Duration::from_secs(1)).await;
+    }
+
+    time::sleep(Duration::from_secs(120)).await;
+}

rumqttc/src/eventloop.rs

@@ -317,6 +317,10 @@ pub(crate) async fn socket_connect(
             SocketAddr::V6(_) => TcpSocket::new_v6()?,
         };
 
+        if let Some(nodelay) = network_options.tcp_nodelay {
+            socket.set_nodelay(nodelay)?;
+        }
+
         if let Some(send_buff_size) = network_options.tcp_send_buffer_size {
             socket.set_send_buffer_size(send_buff_size).unwrap();
         }

rumqttc/src/lib.rs

@@ -369,6 +369,7 @@ impl From<ClientConfig> for TlsConfiguration {
 pub struct NetworkOptions {
     tcp_send_buffer_size: Option<u32>,
     tcp_recv_buffer_size: Option<u32>,
+    tcp_nodelay: Option<bool>,
     conn_timeout: u64,
     #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
     bind_device: Option<String>,
@@ -379,12 +380,17 @@ impl NetworkOptions {
         NetworkOptions {
             tcp_send_buffer_size: None,
             tcp_recv_buffer_size: None,
+            tcp_nodelay: None,
             conn_timeout: 5,
             #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
             bind_device: None,
         }
     }
 
+    pub fn set_tcp_nodelay(&mut self, nodelay: bool) {
+        self.tcp_nodelay = Some(nodelay);
+    }
+
     pub fn set_tcp_send_buffer_size(&mut self, size: u32) {
         self.tcp_send_buffer_size = Some(size);
     }