feat(rumqttd): Support for websocket connections (#633)

---------

Co-authored-by: Henil Dedania <[email protected]>

Author: Nickztar

Cargo.lock

@@ -8,7 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
-
+- Support for Websocket connections (#633)
 ### Changed
 
 ### Deprecated

rumqttd/CHANGELOG.md

@@ -22,8 +22,8 @@ tokio-rustls = { version = "0.24", optional = true }
 rustls-webpki = { version = "0.100.1", optional = true }
 tokio-native-tls = { version = "0.3", optional = true }
 rustls-pemfile = { version = "1", optional = true }
-tokio-tungstenite = { version = "0.15.0", optional = true }
-websocket-codec = { version = "0.5.1", optional = true }
+async-tungstenite = { version = "0.22.2", default-features = false, features = ["tokio-runtime"], optional = true }
+ws_stream_tungstenite = { version = "0.10.0", default-features = false, features = ["tokio_io"], optional = true }
 x509-parser = {version= "0.9.2", optional = true}
 futures-util = { version = "0.3.16", optional = true}
 parking_lot = "0.11.2"
@@ -37,9 +37,10 @@ axum = "0.6.4"
 
 [features]
 default = ["use-rustls"]
-use-rustls = ["dep:tokio-rustls", "dep:rustls-webpki", "dep:rustls-pemfile", "dep:x509-parser"]
-use-native-tls = ["dep:tokio-native-tls", "dep:x509-parser"]
-websockets = ["dep:tokio-tungstenite", "dep:websocket-codec", "dep:tokio-util", "dep:futures-util"]
+use-rustls = ["dep:tokio-rustls", "dep:rustls-webpki", "dep:rustls-pemfile", "dep:x509-parser", "async-tungstenite?/tokio-rustls-native-certs"]
+use-native-tls = ["dep:tokio-native-tls", "dep:x509-parser", "async-tungstenite?/tokio-native-tls"]
+websocket = ["dep:async-tungstenite", "dep:tokio-util", "dep:futures-util", "dep:ws_stream_tungstenite"]
+websockets = ["websocket"]
 validate-tenant-prefix = []
 allow-duplicate-clientid = []
 

rumqttd/Cargo.toml

@@ -82,7 +82,33 @@ next_connection_delay_ms = 1
 listen = "127.0.0.1:9042"
 interval = 1
 
-[ws]
+[ws.1]
+name = "ws-1"
+listen = "0.0.0.0:8083"
+next_connection_delay_ms = 1
+    [ws.1.connections]
+    connection_timeout_ms = 60000
+    max_client_id_len = 256
+    throttle_delay_ms = 0
+    max_payload_size = 20480
+    max_inflight_count = 500
+    max_inflight_size = 1024
+
+# [ws.2]
+# name = "ws-2"
+# listen = "0.0.0.0:8081"
+# next_connection_delay_ms = 1
+#     [ws.2.tls]
+#     capath = "/etc/tls/ca.cert.pem"
+#     certpath = "/etc/tls/server.cert.pem"
+#     keypath = "/etc/tls/server.key.pem"
+#     [ws.2.connections]
+#     connection_timeout_ms = 60000
+#     max_client_id_len = 256
+#     throttle_delay_ms = 0
+#     max_payload_size = 20480
+#     max_inflight_count = 500
+#     max_inflight_size = 1024
 
 [console]
 listen = "0.0.0.0:3030"

rumqttd/rumqttd.toml

@@ -5,6 +5,4 @@ pub mod local;
 pub mod meters;
 pub mod network;
 pub mod remote;
-#[cfg(feature = "websockets")]
-pub mod shadow;
 pub mod timer;

rumqttd/src/link/mod.rs

@@ -142,5 +142,5 @@ impl<P: Protocol> Network<P> {
     }
 }
 
-pub trait N: AsyncRead + AsyncWrite + Send + Sync + Unpin {}
-impl<T> N for T where T: AsyncRead + AsyncWrite + Unpin + Send + Sync {}
+pub trait N: AsyncRead + AsyncWrite + Send + Unpin {}
+impl<T> N for T where T: AsyncRead + AsyncWrite + Unpin + Send {}