From e8eeb0c0768bdb74372188e945d4eb7211af467a Mon Sep 17 00:00:00 2001 From: Michel Heily Date: Thu, 19 Feb 2026 14:53:35 +0200 Subject: [PATCH] feat(rumqttc): update async-tungstenite to 0.32, remove ws_stream_tungstenite async-tungstenite 0.32 includes a built-in bytes module with ByteReader and ByteWriter types that provide tokio::io::AsyncRead/AsyncWrite over WebSocket streams. This makes the ws_stream_tungstenite crate redundant, and that crate pins async-tungstenite ^0.29, blocking the upgrade. Changes: - Update async-tungstenite from 0.29 to 0.32 - Remove ws_stream_tungstenite dependency - Add a custom WsStream adapter in websockets.rs that wraps async-tungstenite's ByteReader/ByteWriter to provide tokio IO traits - Add futures-io and pin-project-lite as optional websocket dependencies - Update v4 and v5 eventloop imports accordingly --- Cargo.lock | 37 +++++------------ rumqttc/Cargo.toml | 7 ++-- rumqttc/src/eventloop.rs | 3 +- rumqttc/src/v5/eventloop.rs | 3 +- rumqttc/src/websockets.rs | 80 +++++++++++++++++++++++++++++++++++++ 5 files changed, 95 insertions(+), 35 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f4f1ec1de..3dee40efc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -200,9 +200,9 @@ dependencies = [ [[package]] name = "async-tungstenite" -version = "0.29.1" +version = "0.32.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef0f7efedeac57d9b26170f72965ecfd31473ca52ca7a64e925b0b6f5f079886" +checksum = "8acc405d38be14342132609f06f02acaf825ddccfe76c4824a69281e0458ebd4" dependencies = [ "atomic-waker", "futures-core", @@ -215,7 +215,7 @@ dependencies = [ "rustls-pki-types", "tokio", "tokio-rustls 0.26.4", - "tungstenite 0.26.2", + "tungstenite 0.28.0", ] [[package]] @@ -2215,17 +2215,19 @@ name = "rumqttc" version = "0.25.1" dependencies = [ "async-http-proxy", - "async-tungstenite 0.29.1", + "async-tungstenite 0.32.1", "bincode", "bytes", "color-backtrace", "fixedbitset 0.5.7", "flume", + "futures-io", "futures-util", "http 1.3.1", "log", "matches", "native-tls", + "pin-project-lite", "pretty_assertions", "pretty_env_logger", "rustls-native-certs", @@ -2239,7 +2241,6 @@ dependencies = [ "tokio-stream", "tokio-util", "url", - "ws_stream_tungstenite 0.15.0", ] [[package]] @@ -2273,7 +2274,7 @@ dependencies = [ "tracing", "tracing-subscriber", "uuid", - "ws_stream_tungstenite 0.13.0", + "ws_stream_tungstenite", "x509-parser", ] @@ -3129,9 +3130,9 @@ dependencies = [ [[package]] name = "tungstenite" -version = "0.26.2" +version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4793cb5e56680ecbb1d843515b23b6de9a75eb04b66643e256a396d43be33c13" +checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" dependencies = [ "bytes", "data-encoding", @@ -3555,26 +3556,6 @@ dependencies = [ "tungstenite 0.21.0", ] -[[package]] -name = "ws_stream_tungstenite" -version = "0.15.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3c9c55940d22313a53398bfeb9438c5f519de475fa37ed7ff068f8c1ca8eb45" -dependencies = [ - "async-tungstenite 0.29.1", - "async_io_stream", - "bitflags 2.10.0", - "futures-core", - "futures-io", - "futures-sink", - "futures-util", - "pharos", - "rustc_version", - "tokio", - "tracing", - "tungstenite 0.26.2", -] - [[package]] name = "x509-parser" version = "0.15.1" diff --git a/rumqttc/Cargo.toml b/rumqttc/Cargo.toml index ceff2c15d..d137820d1 100644 --- a/rumqttc/Cargo.toml +++ b/rumqttc/Cargo.toml @@ -20,7 +20,7 @@ default = ["use-rustls"] use-rustls = ["use-rustls-no-provider", "tokio-rustls/default"] use-rustls-no-provider = ["dep:tokio-rustls", "dep:rustls-webpki", "dep:rustls-pemfile", "dep:rustls-native-certs"] use-native-tls = ["dep:tokio-native-tls", "dep:native-tls"] -websocket = ["dep:async-tungstenite", "dep:ws_stream_tungstenite", "dep:http"] +websocket = ["dep:async-tungstenite", "dep:http", "dep:futures-io", "dep:pin-project-lite"] proxy = ["dep:async-http-proxy"] [dependencies] @@ -39,9 +39,10 @@ rustls-webpki = { version = "0.102.8", optional = true } rustls-pemfile = { version = "2.2.0", optional = true } rustls-native-certs = { version = "0.8.1", optional = true } # websockets -async-tungstenite = { version = "0.29.0", default-features = false, features = ["tokio-rustls-native-certs"], optional = true } -ws_stream_tungstenite = { version= "0.15.0", default-features = false, features = ["tokio_io"], optional = true } +async-tungstenite = { version = "0.32.0", default-features = false, features = ["tokio-rustls-native-certs", "futures-03-sink"], optional = true } http = { version = "1.0.0", optional = true } +pin-project-lite = { version = "0.2", optional = true } +futures-io = { version = "0.3", optional = true } # native-tls tokio-native-tls = { version = "0.3.1", optional = true } native-tls = { version = "0.2.12", optional = true } diff --git a/rumqttc/src/eventloop.rs b/rumqttc/src/eventloop.rs index c2d46f905..9ae876beb 100644 --- a/rumqttc/src/eventloop.rs +++ b/rumqttc/src/eventloop.rs @@ -23,9 +23,8 @@ use crate::tls; #[cfg(feature = "websocket")] use { - crate::websockets::{split_url, validate_response_headers, UrlError}, + crate::websockets::{split_url, validate_response_headers, UrlError, WsStream}, async_tungstenite::tungstenite::client::IntoClientRequest, - ws_stream_tungstenite::WsStream, }; #[cfg(feature = "proxy")] diff --git a/rumqttc/src/v5/eventloop.rs b/rumqttc/src/v5/eventloop.rs index 8fe6f22a5..e75f6479c 100644 --- a/rumqttc/src/v5/eventloop.rs +++ b/rumqttc/src/v5/eventloop.rs @@ -23,9 +23,8 @@ use {std::path::Path, tokio::net::UnixStream}; #[cfg(feature = "websocket")] use { - crate::websockets::{split_url, validate_response_headers, UrlError}, + crate::websockets::{split_url, validate_response_headers, UrlError, WsStream}, async_tungstenite::tungstenite::client::IntoClientRequest, - ws_stream_tungstenite::WsStream, }; #[cfg(feature = "proxy")] diff --git a/rumqttc/src/websockets.rs b/rumqttc/src/websockets.rs index a4dafd51d..328130d1f 100644 --- a/rumqttc/src/websockets.rs +++ b/rumqttc/src/websockets.rs @@ -1,4 +1,14 @@ +use std::{pin::Pin, task::Context}; + +use async_tungstenite::{ + bytes::Sender, + tungstenite::{Error, Message}, + ByteReader, ByteWriter, WebSocketReceiver, WebSocketSender, WebSocketStream, +}; +use futures_util::Stream; use http::{header::ToStrError, Response}; +use pin_project_lite::pin_project; +use tokio::io::{AsyncRead, AsyncWrite}; #[derive(Debug, thiserror::Error)] pub enum UrlError { @@ -71,3 +81,73 @@ fn port(uri: &http::Uri) -> Option { _ => None, }) } + +pin_project! { + +/// Takes a [`WebSocketStream`] and makes it into a byte IO stream +/// compatible with the rest of rumqttc. +pub(crate) struct WsStream { + #[pin] + read_half: ByteReader>, + #[pin] + write_half: ByteWriter>, +} +} + +impl WsStream +where + S: Unpin + futures_io::AsyncWrite + futures_io::AsyncRead, +{ + pub fn new(stream: WebSocketStream) -> Self { + let (sender, receiver) = stream.split(); + + Self { + read_half: ByteReader::new(receiver), + write_half: ByteWriter::new(sender), + } + } +} + +impl AsyncRead for WsStream +where + WebSocketReceiver: Stream> + Unpin, +{ + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + let this = self.project(); + this.read_half.poll_read(cx, buf) + } +} + +impl AsyncWrite for WsStream +where + WebSocketSender: Sender + Unpin, +{ + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + let this = self.project(); + this.write_half.poll_write(cx, buf) + } + + fn poll_flush( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> std::task::Poll> { + let this = self.project(); + this.write_half.poll_flush(cx) + } + + fn poll_shutdown( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> std::task::Poll> { + let this = self.project(); + this.write_half.poll_shutdown(cx) + } +}