open one websocket connection per connection attempt

CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **`WebSocketConfig::with_tls_auto`, and the `with_ca_cert_*` and `with_client_auth_*` methods that call it, now accept a host name URL** (#190). They parsed the URL's host and port as a socket address, so they failed with "Invalid host/port combination" for anything but an IP address. A wss:// connection takes its server and certificate name from the URL and does not use `TlsConfig::addr`, so for a host name that is now the unspecified address `0.0.0.0` with the URL's port; set a real address before using such a config with `TlsTransport`.
 - **`MqttClient` now uses its stored TLS configuration for `wss://` connections** (#190), as it already did for `mqtts://`. A CA, client certificate or key set with `set_tls_config` was ignored over WebSocket, and `set_insecure_tls(true)` replaced the stored configuration with a bare one that disables verification.
 - **The WebSocket client transport now sends its configured headers, subprotocols and user agent** (#165). `WebSocketTransport::connect` built the upgrade request from a fixed set of headers and always offered `Sec-WebSocket-Protocol: mqtt`, so anything set with `WebSocketConfig::with_header`, `with_subprotocol`, `with_subprotocols` or `with_user_agent` never reached the server, and brokers that authenticate the upgrade request through custom headers rejected the connection. The request now carries every custom header, the configured subprotocols in order, followed by `mqtt` when the list does not include it (MQTT-6.0.0-3), and the user agent when one is set. The default user agent is now `mqtt5/<crate version>` instead of `mqtt-v5/0.4.0`.
+- **`MqttClient` now opens one WebSocket connection per connection attempt.** It repeated the whole ws:// or wss:// handshake once for every address the host name resolved to, while each handshake dialed the URL's host itself, so a refused or failed attempt to `localhost` reached the server twice.
 - **The upgrade request's `Host` header now includes a non-default port**, so `ws://broker:8080/mqtt` sends `Host: broker:8080` rather than `Host: broker`.
 
 ### Changed

crates/mqtt5/src/client/inner.rs

@@ -195,10 +195,12 @@ impl MqttClient {
     pub(crate) fn select_addresses_for_connection<'a>(
         addrs: &'a [std::net::SocketAddr],
         host: &str,
+        transport_type: &ClientTransportType,
     ) -> &'a [std::net::SocketAddr] {
-        let is_aws_iot = Self::is_aws_iot_endpoint(host);
-
-        if is_aws_iot {
+        if transport_type.dials_by_url() {
+            tracing::debug!("WebSocket transport dials the URL host, using one attempt");
+            &addrs[0..1]
+        } else if Self::is_aws_iot_endpoint(host) {
             tracing::debug!("AWS IoT endpoint detected, limiting to first resolved address");
             &addrs[0..1]
         } else {
@@ -544,7 +546,8 @@ impl MqttClient {
                 requested_transport_type.redirected()
             };
             let addrs = Self::resolve_addresses(host, port).await?;
-            let addresses_to_try = Self::select_addresses_for_connection(&addrs, host);
+            let addresses_to_try =
+                Self::select_addresses_for_connection(&addrs, host, &client_transport_type);
 
             match self
                 .try_connect_to_addresses(addresses_to_try, client_transport_type, host)

crates/mqtt5/src/client/state.rs

@@ -43,6 +43,14 @@ impl ClientTransportType {
             other => other,
         }
     }
+
+    pub(crate) fn dials_by_url(&self) -> bool {
+        match self {
+            #[cfg(feature = "transport-websocket")]
+            Self::WebSocket(..) | Self::WebSocketSecure(..) => true,
+            _ => false,
+        }
+    }
 }
 
 impl MqttClient {

crates/mqtt5/tests/client_websocket_dial.rs

@@ -0,0 +1,45 @@
+#![cfg(feature = "transport-websocket")]
+
+use mqtt5::time::Duration;
+use mqtt5::{ConnectOptions, MqttClient};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use tokio::net::TcpListener;
+use tokio::time::timeout;
+
+const WAIT: Duration = Duration::from_secs(5);
+
+async fn dials_for_one_failed_connect(scheme: &str) -> usize {
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let port = listener.local_addr().expect("addr").port();
+    let accepted = Arc::new(AtomicUsize::new(0));
+    let counter = Arc::clone(&accepted);
+    tokio::spawn(async move {
+        while let Ok((stream, _)) = listener.accept().await {
+            counter.fetch_add(1, Ordering::SeqCst);
+            drop(stream);
+        }
+    });
+
+    let client = MqttClient::with_options(
+        ConnectOptions::new(format!("dial-{scheme}")).with_automatic_reconnect(false),
+    );
+    let result = timeout(
+        WAIT,
+        client.connect(&format!("{scheme}://localhost:{port}/mqtt")),
+    )
+    .await
+    .expect("connect finishes");
+    assert!(result.is_err(), "connect to a closing server fails");
+    accepted.load(Ordering::SeqCst)
+}
+
+#[tokio::test]
+async fn ws_connect_dials_the_server_once() {
+    assert_eq!(dials_for_one_failed_connect("ws").await, 1);
+}
+
+#[tokio::test]
+async fn wss_connect_dials_the_server_once() {
+    assert_eq!(dials_for_one_failed_connect("wss").await, 1);
+}