feat(mqtt): implement mqtt in wasip2

Signed-off-by: Andrew Steurer <[email protected]>

Author: asteurer

v3/examples/mqtt-outbound/.gitignore

@@ -0,0 +1,2 @@
+main.wasm
+.spin/

v3/examples/mqtt-outbound/README.md

@@ -0,0 +1,11 @@
+# Usage (work in progress)
+
+In one terminal window, run:
+```sh
+docker compose up
+```
+
+In another terminal, you'll run your Spin app:
+```sh
+spin up --build
+```

v3/examples/mqtt-outbound/compose.yaml

@@ -0,0 +1,15 @@
+services:
+  mosquitto:
+    image: eclipse-mosquitto
+    container_name: broker
+    ports:
+      - "1883:1883"
+    command: mosquitto -c /mosquitto-no-auth.conf
+
+  subscriber:
+    image: eclipse-mosquitto
+    container_name: subscriber
+    depends_on:
+      - mosquitto
+    command: mosquitto_sub -h mosquitto -t '#' -v
+    restart: "no" # Clean up container when stopped

v3/examples/mqtt-outbound/go.mod

@@ -0,0 +1,12 @@
+module github.com/http_go
+
+go 1.24
+
+require github.com/spinframework/spin-go-sdk/v3 v3.0.0
+
+require (
+	github.com/julienschmidt/httprouter v1.3.0 // indirect
+	go.bytecodealliance.org/cm v0.2.2 // indirect
+)
+
+replace github.com/spinframework/spin-go-sdk/v3 => ../../

v3/examples/mqtt-outbound/go.sum

@@ -0,0 +1,4 @@
+github.com/julienschmidt/httprouter v1.3.0 h1:U0609e9tgbseu3rBINet9P48AI/D3oJs4dN7jwJOQ1U=
+github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
+go.bytecodealliance.org/cm v0.2.2 h1:M9iHS6qs884mbQbIjtLX1OifgyPG9DuMs2iwz8G4WQA=
+go.bytecodealliance.org/cm v0.2.2/go.mod h1:JD5vtVNZv7sBoQQkvBvAAVKJPhR/bqBH7yYXTItMfZI=

v3/examples/mqtt-outbound/main.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+
+	spinhttp "github.com/spinframework/spin-go-sdk/v3/http"
+	"github.com/spinframework/spin-go-sdk/v3/mqtt"
+)
+
+func main() {}
+
+func init() {
+	spinhttp.Handle(func(w http.ResponseWriter, r *http.Request) {
+		var missingEnvs []string
+		addr := validateEnv("MQTT_ADDRESS", &missingEnvs)
+		usr := validateEnv("MQTT_USERNAME", &missingEnvs)
+		pass := validateEnv("MQTT_PASSWORD", &missingEnvs)
+		keepAliveStr := validateEnv("MQTT_KEEP_ALIVE_INTERVAL", &missingEnvs)
+		topic := validateEnv("MQTT_TOPIC", &missingEnvs)
+
+		if len(missingEnvs) > 0 {
+			panic("Missing the following environment variables:\n" + strings.Join(missingEnvs, "\n"))
+		}
+
+		keepAlive, err := strconv.Atoi(keepAliveStr)
+		if err != nil {
+			panic("MQTT_KEEP_ALIVE_INTERVAL is not valid: must be an integer")
+		}
+
+		conn, err := mqtt.OpenConnection(addr, usr, pass, uint64(keepAlive))
+		if err != nil {
+			panic(err)
+		}
+
+		message := []byte("Eureka!")
+
+		if err := mqtt.Publish(conn, topic, message, 0); err != nil {
+			panic(err)
+		}
+	})
+}
+
+func validateEnv(key string, missingEnvs *[]string) (value string) {
+	value, exists := os.LookupEnv(key)
+	if !exists {
+		if missingEnvs != nil {
+			*missingEnvs = append(*missingEnvs, key)
+		}
+	}
+	return value
+}

v3/examples/mqtt-outbound/spin.toml

@@ -0,0 +1,20 @@
+spin_manifest_version = 2
+
+[application]
+name = "go-mqtt-outbound-example"
+version = "0.1.0"
+authors = ["Andrew Steurer <[email protected]>"]
+description = "Using Spin with MQTT"
+
+[[trigger.http]]
+route = "/publish"
+component = "mqtt-outbound"
+
+[component.mqtt-outbound]
+source = "main.wasm"
+# To test anonymous MQTT authentication, remove the values from MQTT_USERNAME and MQTT_PASSWORD env variables.
+environment = { MQTT_ADDRESS = "mqtt://127.0.0.1:1883?client_id=client001", MQTT_USERNAME = "user", MQTT_PASSWORD = "password", MQTT_KEEP_ALIVE_INTERVAL = "30", MQTT_TOPIC = "telemetry" }
+allowed_outbound_hosts = ["mqtt://127.0.0.1:1883"]
+[component.mqtt-outbound.build]
+command = "tinygo build -target=wasip2 --wit-package $(go list -mod=readonly -m -f '{{.Dir}}' github.com/spinframework/spin-go-sdk/v3)/wit --wit-world http-trigger -gc=leaking -o main.wasm main.go"
+watch = ["**/*.go", "go.mod"]

v3/mqtt/mqtt.go

@@ -0,0 +1,49 @@
+package mqtt
+
+import (
+	"errors"
+
+	"github.com/spinframework/spin-go-sdk/v3/internal/fermyon/spin/v2.0.0/mqtt"
+	"go.bytecodealliance.org/cm"
+)
+
+// OpenConnection initializes an MQTT connection
+func OpenConnection(address, username, password string, keepAliveIntervalInSecs uint64) (mqtt.Connection, error) {
+	conn, err, isErr := mqtt.ConnectionOpen(address, username, password, keepAliveIntervalInSecs).Result()
+	if isErr {
+		return cm.ResourceNone, toError(&err)
+	}
+
+	return conn, nil
+}
+
+// Publish publishes an MQTT message
+func Publish(conn mqtt.Connection, topic string, payload []byte, qos uint8) error {
+	_, err, isErr := conn.Publish(topic, mqtt.Payload(cm.ToList(payload)), toQos(qos)).Result()
+	if isErr {
+		return toError(&err)
+	}
+
+	return nil
+}
+
+func toQos(q uint8) mqtt.Qos {
+	switch q {
+	case 0:
+		return mqtt.QosAtMostOnce
+	case 1:
+		return mqtt.QosAtLeastOnce
+	case 2:
+		return mqtt.QosExactlyOnce
+	default:
+		panic("invalid QoS")
+	}
+}
+
+func toError(err *mqtt.Error) error {
+	if err == nil {
+		return nil
+	}
+
+	return errors.New(err.String())
+}