feat(outputs.mqtt): Add websocket URI support

plugins/common/mqtt/mqtt_v3.go

@@ -71,8 +71,15 @@ func NewMQTTv311Client(cfg *MqttConfig) (*mqttv311Client, error) {
 		return nil, err
 	}
 	for _, server := range servers {
-		if tlsCfg != nil {
-			server.Scheme = "tls"
+		switch server.Scheme {
+		case "ws":
+			// Keep ws:// as-is
+		case "wss":
+			// Keep wss:// as-is
+		default:
+			if tlsCfg != nil {
+				server.Scheme = "tls"
+			}
 		}
 		broker := server.String()
 		opts.AddBroker(broker)

plugins/common/mqtt/mqtt_v5.go

@@ -67,8 +67,13 @@ func NewMQTTv5Client(cfg *MqttConfig) (*mqttv5Client, error) {
 
 	brokers := make([]*url.URL, 0, len(servers))
 	for _, server := range servers {
-		if tlsCfg != nil {
-			server.Scheme = "tls"
+		switch server.Scheme {
+		case "ws", "wss":
+			// Keep websocket schemes as-is
+		default:
+			if tlsCfg != nil {
+				server.Scheme = "tls"
+			}
 		}
 		brokers = append(brokers, server)
 	}

plugins/outputs/mqtt/sample.conf

@@ -4,9 +4,9 @@
   ## The list of brokers should only include the hostname or IP address and the
   ## port to the broker. This should follow the format `[{scheme}://]{host}:{port}`. For
   ## example, `localhost:1883` or `mqtt://localhost:1883`.
-  ## Scheme can be any of the following: tcp://, mqtt://, tls://, mqtts://
+  ## Scheme can be any of the following: tcp://, mqtt://, tls://, mqtts://, ws://, wss://
   ## non-TLS and TLS servers can not be mix-and-matched.
-  servers = ["localhost:1883", ] # or ["mqtts://tls.example.com:1883"]
+  servers = ["localhost:1883", ] # or ["mqtts://tls.example.com:1883"] or ["wss://broker.example.com:443"]
 
   ## Protocol can be `3.1.1` or `5`. Default is `3.1.1`
   # protocol = "3.1.1"