Support websocket transport for MQTT.

This also includes a workaround for an issue in rumqttc where the
`parse_url` function did not properly handle websocket URLs.

Author: Syfaro

Cargo.lock

@@ -30,7 +30,7 @@ reqwest = { version = "0.12.5", default-features = false, features = [
     "stream",
 ] }
 rhai = { version = "1.19.0", features = ["serde", "sync"] }
-rumqttc = { version = "0.24.0", features = ["url"] }
+rumqttc = { version = "0.24.0", features = ["url", "websocket"] }
 serde = { version = "1.0.204", features = ["derive", "rc"] }
 serde_json = "1.0.120"
 serde_with = "3.8.3"

Cargo.toml

@@ -50,13 +50,34 @@ pub struct WebSocketConnectionConfig {
 
 #[derive(Debug, Deserialize)]
 pub struct MqttConnectionConfig {
-    url: String,
+    #[serde(flatten)]
+    params: MqttConnectionParams,
     publish_topic: String,
     action_topic: Option<String>,
     #[serde(default)]
     allow_actions: bool,
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(untagged)]
+pub enum MqttConnectionParams {
+    Url {
+        url: String,
+    },
+    Options {
+        client_id: String,
+        host: String,
+        port: Option<u16>,
+        credentials: Option<MqttCredentials>,
+    },
+}
+
+#[derive(Debug, Deserialize)]
+pub struct MqttCredentials {
+    username: String,
+    password: String,
+}
+
 #[derive(Debug, Deserialize)]
 #[serde(tag = "action", rename_all = "snake_case")]
 pub enum ConnectionAction {

src/connection/mod.rs

@@ -1,6 +1,7 @@
 use std::collections::{HashMap, HashSet};
 
 use async_trait::async_trait;
+use eyre::{bail, OptionExt};
 use tokio::{
     select,
     sync::mpsc::{channel, Sender},
@@ -23,7 +24,48 @@ impl MqttConnection {
         action_tx: Sender<super::ConnectionAction>,
     ) -> eyre::Result<Self> {
         let (tx, mut rx) = channel::<(serde_json::Value, HashMap<String, String>)>(1);
-        let options = rumqttc::MqttOptions::parse_url(config.url)?;
+
+        let options = match config.params {
+            super::MqttConnectionParams::Url { url } => rumqttc::MqttOptions::parse_url(url)?,
+            super::MqttConnectionParams::Options {
+                client_id,
+                host,
+                port,
+                credentials,
+            } => {
+                let mut options = if let Ok(url) = url::Url::parse(&host) {
+                    let (transport, host) = match url.scheme() {
+                        "mqtt" => (
+                            rumqttc::Transport::tcp(),
+                            url.host_str().ok_or_eyre("mqtt url scheme missing host")?,
+                        ),
+                        "mqtts" => (
+                            rumqttc::Transport::tls_with_default_config(),
+                            url.host_str().ok_or_eyre("mqtts url scheme missing host")?,
+                        ),
+                        "ws" => (rumqttc::Transport::ws(), url.as_str()),
+                        "wss" => (rumqttc::Transport::wss_with_default_config(), url.as_str()),
+                        scheme => bail!("unknown scheme: {scheme}"),
+                    };
+
+                    let mut options =
+                        rumqttc::MqttOptions::new(client_id, host, url.port().unwrap_or(1883));
+
+                    options.set_transport(transport);
+
+                    options
+                } else {
+                    rumqttc::MqttOptions::new(client_id, host, port.unwrap_or(1883))
+                };
+
+                if let Some(super::MqttCredentials { username, password }) = credentials {
+                    options.set_credentials(username, password);
+                }
+
+                options
+            }
+        };
+
         let (client, mut event_loop) = rumqttc::AsyncClient::new(options, 1);
 
         if let Some(action_topic) = config.action_topic.as_deref() {