Prototype a MQTT Source

cmd/mqttsource/main.go

@@ -0,0 +1,131 @@
+/*
+Copyright 2024 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"net"
+	"os"
+
+	mqttv2 "github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2"
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/eclipse/paho.golang/paho"
+)
+
+var (
+	sink   string
+	source string
+
+	// CloudEvents specific parameters
+	eventSource string
+
+	topic    string
+	clientid string
+)
+
+func init() {
+	flag.StringVar(&sink, "sink", "", "the host url to send messages to")
+	flag.StringVar(&source, "source", "", "the url to get messages from")
+	flag.StringVar(&eventSource, "eventSource", "", "the event-source (CloudEvents)")
+
+	flag.StringVar(&topic, "topic", "mqtt-topic", "MQTT topic subscribe to")
+	flag.StringVar(&clientid, "clientid", "receiver-client-id", "MQTT source client id")
+}
+
+func main() {
+	flag.Parse()
+
+	kSink := os.Getenv("K_SINK")
+	if kSink != "" {
+		sink = kSink
+	}
+
+	// "source" flag must not be empty for operation.
+	if source == "" {
+		log.Fatal("A valid MQTT broker URL must be defined.")
+	}
+
+	// The event's source defaults to the MQTT broker URL.
+	if eventSource == "" {
+		eventSource = source
+	}
+
+	ctx := cloudevents.ContextWithTarget(context.Background(), sink)
+
+	conn, err := net.Dial("tcp", source)
+	if err != nil {
+		log.Fatalf("failed to connect to MQTT broker: %s", err.Error())
+	}
+
+	config := &paho.ClientConfig{
+		ClientID: clientid,
+		Conn:     conn,
+	}
+
+	subscribeOpt := &paho.Subscribe{
+		Subscriptions: []paho.SubscribeOptions{
+			{Topic: topic,
+				QoS: 0},
+		},
+	}
+
+	mqttReceiver, err := mqttv2.New(ctx, config, mqttv2.WithSubscribe(subscribeOpt))
+	if err != nil {
+		log.Fatalf("failed to create protocol: %s", err.Error())
+	}
+	defer mqttReceiver.Close(ctx)
+
+	ceReceiver, err := cloudevents.NewClient(mqttReceiver)
+	if err != nil {
+		log.Fatalf("failed to create client: %v", err)
+	}
+
+	ceClient, err := cloudevents.NewClientHTTP()
+	if err != nil {
+		log.Fatalf("Failed to create a http cloudevent client: %s", err.Error())
+	}
+
+	log.Printf("MQTT source start consuming messages from %s\n", source)
+	err = ceReceiver.StartReceiver(ctx, func(ctx context.Context, event cloudevents.Event) {
+		receive(ctx, event, ceClient)
+	})
+	if err != nil {
+		log.Fatalf("failed to start receiver: %s", err)
+	} else {
+		log.Printf("MQTT source stopped\n")
+	}
+
+}
+
+func receive(ctx context.Context, event cloudevents.Event, c cloudevents.Client) {
+	log.Printf("Received event: %s", event)
+	data := event.Data()
+	newEvent := cloudevents.NewEvent(cloudevents.VersionV1)
+	newEvent.SetType(event.Type())
+	newEvent.SetSource(eventSource)
+	newEvent.SetID(event.ID())
+	err := newEvent.SetData(cloudevents.ApplicationJSON, data)
+	if err != nil {
+		log.Printf("Error setting event data: %v", err)
+		return
+	}
+	if result := c.Send(ctx, newEvent); !cloudevents.IsACK(result) {
+		log.Printf("Sending event to sink failed: %v", result)
+	}
+}

config/tools/mqttsource/mqttsource.yaml

@@ -0,0 +1,40 @@
+# Copyright 2021 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This is a very simple Knative Eventing Source that receive messages
+# from MQTT brokers and send to sink as Cloudevents
+apiVersion: sources.knative.dev/v1
+kind: ContainerSource
+metadata:
+  name: mqttsource
+spec:
+  template:
+    spec:
+      containers:
+        - image: ko://knative.dev/eventing/cmd/mqttsource
+          securityContext:
+            allowPrivilegeEscalation: false
+            readOnlyRootFilesystem: true
+            runAsNonRoot: true
+            capabilities:
+              drop:
+                - ALL
+            seccompProfile:
+              type: RuntimeDefault
+
+  sink:
+    ref:
+      apiVersion: serving.knative.dev/v1
+      kind: Service
+      name: event-display

go.mod

@@ -6,9 +6,11 @@ require (
 	github.com/ahmetb/gen-crd-api-reference-docs v0.3.1-0.20210420163308-c1402a70e2f1
 	github.com/cloudevents/conformance v0.2.0
 	github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.15.2
+	github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2 v2.0.0-20240508060731-1ed9471c98bd
 	github.com/cloudevents/sdk-go/sql/v2 v2.15.2
 	github.com/cloudevents/sdk-go/v2 v2.15.2
 	github.com/coreos/go-oidc/v3 v3.9.0
+	github.com/eclipse/paho.golang v0.12.0
 	github.com/go-jose/go-jose/v3 v3.0.3
 	github.com/golang/protobuf v1.5.4
 	github.com/google/go-cmp v0.6.0

go.sum

@@ -90,6 +90,8 @@ github.com/cloudevents/conformance v0.2.0 h1:NvSXOKlagcsOWMEbi8U7Ex/0oQ4JZE1HQ45
 github.com/cloudevents/conformance v0.2.0/go.mod h1:rHKDwylBH89Rns6U3wL9ww8bg9/4GbwRCDNuyoC6bcc=
 github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.15.2 h1:AbtPqiUDzKup5JpTZzO297/QXgL/TAdpdXQCNwLzlaM=
 github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.15.2/go.mod h1:ZbYLE+yaEQ2j4vbRc9qzvGmg30A9LhwFt/1bSebNnbU=
+github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2 v2.0.0-20240508060731-1ed9471c98bd h1:MGVlnkCz/b0vjfkM5tSVLD+4oaUnYuVEjiW6lAMJ9IM=
+github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2 v2.0.0-20240508060731-1ed9471c98bd/go.mod h1:s+KZsVZst0bVW6vuKYb8CH49CcSJDO09+ZiIeKzJmqE=
 github.com/cloudevents/sdk-go/sql/v2 v2.15.2 h1:TNaTeWIbDaci89xgXbmmNVGccawQOvEfWYLWrr7Fk/k=
 github.com/cloudevents/sdk-go/sql/v2 v2.15.2/go.mod h1:us+PSk8OXdk8pDbRfvxy5w8ub5goKE7UP9PjKDY7TPw=
 github.com/cloudevents/sdk-go/v2 v2.15.2 h1:54+I5xQEnI73RBhWHxbI1XJcqOFOVJN85vb41+8mHUc=
@@ -111,6 +113,8 @@ github.com/dgryski/go-gk v0.0.0-20200319235926-a69029f61654/go.mod h1:qm+vckxRlD
 github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
+github.com/eclipse/paho.golang v0.12.0 h1:EXQFJbJklDnUqW6lyAknMWRhM2NgpHxwrrL8riUmp3Q=
+github.com/eclipse/paho.golang v0.12.0/go.mod h1:TSDCUivu9JnoR9Hl+H7sQMcHkejWH2/xKK1NJGtLbIE=
 github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
 github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/envoyproxy/go-control-plane v0.6.9/go.mod h1:SBwIajubJHhxtWwsL9s8ss4safvEdbitLhGGK48rN6g=