mqtt transport v2

Signed-off-by: Wei Liu <[email protected]>

Author: skeeey

pkg/cloudevents/generic/options/mqtt/agentoptions.go

@@ -125,11 +125,16 @@ func AgentPubTopic(ctx context.Context, o *MQTTOptions, clusterName string, evtC
 		return "", fmt.Errorf("unsupported event type %q, %v", ceType, err)
 	}
 
-	originalSource, err := evtCtx.GetExtension(types.ExtensionOriginalSource)
+	originalSourceVal, err := evtCtx.GetExtension(types.ExtensionOriginalSource)
 	if err != nil {
 		return "", err
 	}
 
+	originalSource, ok := originalSourceVal.(string)
+	if !ok {
+		return "", fmt.Errorf("originalsource extension must be a string, got %T", originalSourceVal)
+	}
+
 	// agent request to sync resource spec from all sources
 	if eventType.Action == types.ResyncRequestAction && originalSource == types.SourceAll {
 		if len(o.Topics.AgentBroadcast) == 0 {

pkg/cloudevents/generic/options/mqtt/options.go

@@ -4,13 +4,14 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
-	"k8s.io/klog/v2"
 	"net"
 	"os"
 	"regexp"
 	"strings"
 	"time"
 
+	"k8s.io/klog/v2"
+
 	cloudeventsmqtt "github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2"
 	"github.com/eclipse/paho.golang/packets"
 	"github.com/eclipse/paho.golang/paho"
@@ -235,8 +236,8 @@ func (o *MQTTOptions) GetCloudEventsProtocol(
 
 	opts := []cloudeventsmqtt.Option{
 		cloudeventsmqtt.WithConnect(o.GetMQTTConnectOption(clientID)),
-		cloudeventsmqtt.WithDebugLogger(&PahoDebugLogger{logger: logger}),
-		cloudeventsmqtt.WithErrorLogger(&PahoErrorLogger{logger: logger}),
+		cloudeventsmqtt.WithDebugLogger(NewPahoDebugLogger(logger)),
+		cloudeventsmqtt.WithErrorLogger(NewPahoErrorLogger(logger)),
 	}
 	opts = append(opts, clientOpts...)
 	return cloudeventsmqtt.New(ctx, config, opts...)

pkg/cloudevents/generic/options/mqtt/sourceoptions.go

@@ -122,11 +122,16 @@ func SourcePubTopic(ctx context.Context, o *MQTTOptions, sourceID string, evtCtx
 		return "", fmt.Errorf("unsupported event type %q, %v", ceType, err)
 	}
 
-	clusterName, err := evtCtx.GetExtension(types.ExtensionClusterName)
+	clusterNameVal, err := evtCtx.GetExtension(types.ExtensionClusterName)
 	if err != nil {
 		return "", err
 	}
 
+	clusterName, ok := clusterNameVal.(string)
+	if !ok {
+		return "", fmt.Errorf("clustername extension must be a string, got %T", clusterNameVal)
+	}
+
 	if eventType.Action == types.ResyncRequestAction && clusterName == types.ClusterAll {
 		// source request to get resources status from all agents
 		if len(o.Topics.SourceBroadcast) == 0 {
@@ -138,7 +143,7 @@ func SourcePubTopic(ctx context.Context, o *MQTTOptions, sourceID string, evtCtx
 	}
 
 	// source publishes spec events or status resync events
-	eventsTopic := strings.Replace(o.Topics.SourceEvents, "+", fmt.Sprintf("%s", clusterName), 1)
+	eventsTopic := strings.Replace(o.Topics.SourceEvents, "+", clusterName, 1)
 	return eventsTopic, nil
 }
 
@@ -149,8 +154,8 @@ func SourceSubscribe(o *MQTTOptions, sourceID string) (*paho.Subscribe, error) {
 	}
 
 	if topicSource != sourceID {
-		return nil, fmt.Errorf("the topic source %q does not match with the client sourceID %q",
-			o.Topics.AgentEvents, sourceID)
+		return nil, fmt.Errorf("the topic source %q does not match the client sourceID %q",
+			topicSource, sourceID)
 	}
 
 	subscribe := &paho.Subscribe{