mqtt transport v2

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

Author: skeeey

pkg/cloudevents/generic/options/builder/optionsbuilder.go

@@ -8,6 +8,7 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
 	grpcv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/grpc"
+	mqttv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/mqtt"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 )
 
@@ -56,7 +57,7 @@ func BuildCloudEventsSourceOptions(config any,
 	clientId, sourceId string, dataType types.CloudEventsDataType) (*options.CloudEventsSourceOptions, error) {
 	switch config := config.(type) {
 	case *mqtt.MQTTOptions:
-		return mqtt.NewSourceOptions(config, clientId, sourceId), nil
+		return mqttv2.NewSourceOptions(config, clientId, sourceId), nil
 	case *grpc.GRPCOptions:
 		return grpcv2.NewSourceOptions(config, sourceId, dataType), nil
 	default:
@@ -69,7 +70,7 @@ func BuildCloudEventsAgentOptions(config any,
 	clusterName, clientId string, dataType types.CloudEventsDataType) (*options.CloudEventsAgentOptions, error) {
 	switch config := config.(type) {
 	case *mqtt.MQTTOptions:
-		return mqtt.NewAgentOptions(config, clusterName, clientId), nil
+		return mqttv2.NewAgentOptions(config, clusterName, clientId), nil
 	case *grpc.GRPCOptions:
 		return grpcv2.NewAgentOptions(config, clusterName, clientId, dataType), nil
 	default:

pkg/cloudevents/generic/options/builder/optionsbuilder_test.go

@@ -56,7 +56,7 @@ func TestBuildCloudEventsSourceOptions(t *testing.T) {
 					Timeout:    60 * time.Second,
 				},
 			},
-			expectedTransportType: "*mqtt.mqttSourceTransport",
+			expectedTransportType: "*mqtt.mqttTransport",
 		},
 		{
 			name:       "grpc config",

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

@@ -24,6 +24,7 @@ type mqttAgentTransport struct {
 	agentID           string
 }
 
+// Deprecated: use v2.mqtt.NewAgentOptions instead
 func NewAgentOptions(mqttOptions *MQTTOptions, clusterName, agentID string) *options.CloudEventsAgentOptions {
 	mqttAgentOptions := &mqttAgentTransport{
 		MQTTOptions: *mqttOptions,
@@ -40,8 +41,6 @@ func NewAgentOptions(mqttOptions *MQTTOptions, clusterName, agentID string) *opt
 }
 
 func (o *mqttAgentTransport) WithContext(ctx context.Context, evtCtx cloudevents.EventContext) (context.Context, error) {
-	logger := klog.FromContext(ctx)
-
 	topic, err := getAgentPubTopic(ctx)
 	if err != nil {
 		return nil, err
@@ -51,59 +50,18 @@ func (o *mqttAgentTransport) WithContext(ctx context.Context, evtCtx cloudevents
 		return cloudeventscontext.WithTopic(ctx, string(*topic)), nil
 	}
 
-	eventType, err := types.ParseCloudEventsType(evtCtx.GetType())
-	if err != nil {
-		return nil, fmt.Errorf("unsupported event type %s, %v", eventType, err)
-	}
-
-	originalSource, err := evtCtx.GetExtension(types.ExtensionOriginalSource)
-	if err != nil {
-		return nil, err
-	}
-
-	// agent request to sync resource spec from all sources
-	if eventType.Action == types.ResyncRequestAction && originalSource == types.SourceAll {
-		if len(o.Topics.AgentBroadcast) == 0 {
-			logger.Info("the agent broadcast topic not set, fall back to the agent events topic")
-
-			// TODO after supporting multiple sources, we should list each source
-			eventsTopic := replaceLast(o.Topics.AgentEvents, "+", o.clusterName)
-			return cloudeventscontext.WithTopic(ctx, eventsTopic), nil
-		}
-
-		resyncTopic := strings.Replace(o.Topics.AgentBroadcast, "+", o.clusterName, 1)
-		return cloudeventscontext.WithTopic(ctx, resyncTopic), nil
-	}
-
-	topicSource, err := getSourceFromEventsTopic(o.Topics.AgentEvents)
+	pubTopic, err := AgentPubTopic(ctx, &o.MQTTOptions, o.clusterName, evtCtx)
 	if err != nil {
 		return nil, err
 	}
 
-	// agent publishes status events or spec resync events
-	eventsTopic := replaceLast(o.Topics.AgentEvents, "+", o.clusterName)
-	eventsTopic = replaceLast(eventsTopic, "+", topicSource)
-	return cloudeventscontext.WithTopic(ctx, eventsTopic), nil
+	return cloudeventscontext.WithTopic(ctx, pubTopic), nil
 }
 
 func (o *mqttAgentTransport) Connect(ctx context.Context) error {
-	subscribe := &paho.Subscribe{
-		Subscriptions: []paho.SubscribeOptions{
-			{
-				// TODO support multiple sources, currently the client require the source events topic has a sourceID, in
-				// the future, client may need a source list, it will subscribe to each source
-				// receiving the sources events
-				Topic: replaceLast(o.Topics.SourceEvents, "+", o.clusterName), QoS: byte(o.SubQoS),
-			},
-		},
-	}
-
-	// receiving status resync events from all sources
-	if len(o.Topics.SourceBroadcast) != 0 {
-		subscribe.Subscriptions = append(subscribe.Subscriptions, paho.SubscribeOptions{
-			Topic: o.Topics.SourceBroadcast,
-			QoS:   byte(o.SubQoS),
-		})
+	subscribe, err := AgentSubscribe(&o.MQTTOptions, o.clusterName)
+	if err != nil {
+		return err
 	}
 
 	protocol, err := o.GetCloudEventsProtocol(
@@ -157,3 +115,62 @@ func (o *mqttAgentTransport) Close(ctx context.Context) error {
 func (o *mqttAgentTransport) ErrorChan() <-chan error {
 	return o.errorChan
 }
+
+func AgentPubTopic(ctx context.Context, o *MQTTOptions, clusterName string, evtCtx cloudevents.EventContext) (string, error) {
+	logger := klog.FromContext(ctx)
+
+	ceType := evtCtx.GetType()
+	eventType, err := types.ParseCloudEventsType(ceType)
+	if err != nil {
+		return "", fmt.Errorf("unsupported event type %q, %v", ceType, err)
+	}
+
+	originalSource, err := evtCtx.GetExtension(types.ExtensionOriginalSource)
+	if err != nil {
+		return "", err
+	}
+
+	// agent request to sync resource spec from all sources
+	if eventType.Action == types.ResyncRequestAction && originalSource == types.SourceAll {
+		if len(o.Topics.AgentBroadcast) == 0 {
+			logger.Info("the agent broadcast topic not set, fall back to the agent events topic")
+
+			// TODO after supporting multiple sources, we should list each source
+			return replaceLast(o.Topics.AgentEvents, "+", clusterName), nil
+		}
+
+		return strings.Replace(o.Topics.AgentBroadcast, "+", clusterName, 1), nil
+	}
+
+	topicSource, err := getSourceFromEventsTopic(o.Topics.AgentEvents)
+	if err != nil {
+		return "", err
+	}
+
+	// agent publishes status events or spec resync events
+	eventsTopic := replaceLast(o.Topics.AgentEvents, "+", clusterName)
+	return replaceLast(eventsTopic, "+", topicSource), nil
+}
+
+func AgentSubscribe(o *MQTTOptions, clusterName string) (*paho.Subscribe, error) {
+	subscribe := &paho.Subscribe{
+		Subscriptions: []paho.SubscribeOptions{
+			{
+				// TODO support multiple sources, currently the client require the source events topic has a sourceID, in
+				// the future, client may need a source list, it will subscribe to each source
+				// receiving the sources events
+				Topic: replaceLast(o.Topics.SourceEvents, "+", clusterName), QoS: byte(o.SubQoS),
+			},
+		},
+	}
+
+	// receiving status resync events from all sources
+	if len(o.Topics.SourceBroadcast) != 0 {
+		subscribe.Subscriptions = append(subscribe.Subscriptions, paho.SubscribeOptions{
+			Topic: o.Topics.SourceBroadcast,
+			QoS:   byte(o.SubQoS),
+		})
+	}
+
+	return subscribe, nil
+}

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

@@ -2,6 +2,7 @@ package mqtt
 
 import (
 	"fmt"
+
 	"github.com/eclipse/paho.golang/paho/log"
 	"k8s.io/klog/v2"
 )
@@ -14,8 +15,13 @@ type PahoDebugLogger struct {
 	logger klog.Logger
 }
 
-var _ log.Logger = &PahoErrorLogger{}
-var _ log.Logger = &PahoDebugLogger{}
+func NewPahoErrorLogger(logger klog.Logger) log.Logger {
+	return &PahoErrorLogger{logger: logger}
+}
+
+func NewPahoDebugLogger(logger klog.Logger) log.Logger {
+	return &PahoDebugLogger{logger: logger}
+}
 
 func (l *PahoErrorLogger) Println(v ...interface{}) {
 	l.logger.Error(fmt.Errorf("get err %s", fmt.Sprint(v...)), "MQTT error message")

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

@@ -23,6 +23,7 @@ type mqttSourceTransport struct {
 	clientID          string
 }
 
+// Deprecated: use v2.mqtt.NewSourceOptions instead
 func NewSourceOptions(mqttOptions *MQTTOptions, clientID, sourceID string) *options.CloudEventsSourceOptions {
 	mqttSourceOptions := &mqttSourceTransport{
 		MQTTOptions: *mqttOptions,
@@ -47,58 +48,20 @@ func (o *mqttSourceTransport) WithContext(ctx context.Context, evtCtx cloudevent
 		return cloudeventscontext.WithTopic(ctx, string(*topic)), nil
 	}
 
-	eventType, err := types.ParseCloudEventsType(evtCtx.GetType())
-	if err != nil {
-		return nil, fmt.Errorf("unsupported event type %s, %v", eventType, err)
-	}
-
-	clusterName, err := evtCtx.GetExtension(types.ExtensionClusterName)
+	pubTopic, err := SourcePubTopic(ctx, &o.MQTTOptions, o.sourceID, evtCtx)
 	if err != nil {
 		return nil, err
 	}
 
-	if eventType.Action == types.ResyncRequestAction && clusterName == types.ClusterAll {
-		// source request to get resources status from all agents
-		if len(o.Topics.SourceBroadcast) == 0 {
-			return nil, fmt.Errorf("the source broadcast topic not set")
-		}
-
-		resyncTopic := strings.Replace(o.Topics.SourceBroadcast, "+", o.sourceID, 1)
-		return cloudeventscontext.WithTopic(ctx, resyncTopic), nil
-	}
-
-	// source publishes spec events or status resync events
-	eventsTopic := strings.Replace(o.Topics.SourceEvents, "+", fmt.Sprintf("%s", clusterName), 1)
-	return cloudeventscontext.WithTopic(ctx, eventsTopic), nil
+	return cloudeventscontext.WithTopic(ctx, pubTopic), nil
 }
 
 func (o *mqttSourceTransport) Connect(ctx context.Context) error {
-	topicSource, err := getSourceFromEventsTopic(o.Topics.AgentEvents)
+	subscribe, err := SourceSubscribe(&o.MQTTOptions, o.sourceID)
 	if err != nil {
 		return err
 	}
 
-	if topicSource != o.sourceID {
-		return fmt.Errorf("the topic source %q does not match with the client sourceID %q",
-			o.Topics.AgentEvents, o.sourceID)
-	}
-
-	subscribe := &paho.Subscribe{
-		Subscriptions: []paho.SubscribeOptions{
-			{
-				Topic: o.Topics.AgentEvents, QoS: byte(o.SubQoS),
-			},
-		},
-	}
-
-	if len(o.Topics.AgentBroadcast) != 0 {
-		// receiving spec resync events from all agents
-		subscribe.Subscriptions = append(subscribe.Subscriptions, paho.SubscribeOptions{
-			Topic: o.Topics.AgentBroadcast,
-			QoS:   byte(o.SubQoS),
-		})
-	}
-
 	protocol, err := o.GetCloudEventsProtocol(
 		ctx,
 		o.clientID,
@@ -151,3 +114,60 @@ func (o *mqttSourceTransport) Close(ctx context.Context) error {
 func (o *mqttSourceTransport) ErrorChan() <-chan error {
 	return o.errorChan
 }
+
+func SourcePubTopic(ctx context.Context, o *MQTTOptions, sourceID string, evtCtx cloudevents.EventContext) (string, error) {
+	ceType := evtCtx.GetType()
+	eventType, err := types.ParseCloudEventsType(ceType)
+	if err != nil {
+		return "", fmt.Errorf("unsupported event type %q, %v", ceType, err)
+	}
+
+	clusterName, err := evtCtx.GetExtension(types.ExtensionClusterName)
+	if err != nil {
+		return "", err
+	}
+
+	if eventType.Action == types.ResyncRequestAction && clusterName == types.ClusterAll {
+		// source request to get resources status from all agents
+		if len(o.Topics.SourceBroadcast) == 0 {
+			return "", fmt.Errorf("the source broadcast topic not set")
+		}
+
+		resyncTopic := strings.Replace(o.Topics.SourceBroadcast, "+", sourceID, 1)
+		return resyncTopic, nil
+	}
+
+	// source publishes spec events or status resync events
+	eventsTopic := strings.Replace(o.Topics.SourceEvents, "+", fmt.Sprintf("%s", clusterName), 1)
+	return eventsTopic, nil
+}
+
+func SourceSubscribe(o *MQTTOptions, sourceID string) (*paho.Subscribe, error) {
+	topicSource, err := getSourceFromEventsTopic(o.Topics.AgentEvents)
+	if err != nil {
+		return nil, err
+	}
+
+	if topicSource != sourceID {
+		return nil, fmt.Errorf("the topic source %q does not match with the client sourceID %q",
+			o.Topics.AgentEvents, sourceID)
+	}
+
+	subscribe := &paho.Subscribe{
+		Subscriptions: []paho.SubscribeOptions{
+			{
+				Topic: o.Topics.AgentEvents, QoS: byte(o.SubQoS),
+			},
+		},
+	}
+
+	if len(o.Topics.AgentBroadcast) != 0 {
+		// receiving spec resync events from all agents
+		subscribe.Subscriptions = append(subscribe.Subscriptions, paho.SubscribeOptions{
+			Topic: o.Topics.AgentBroadcast,
+			QoS:   byte(o.SubQoS),
+		})
+	}
+
+	return subscribe, nil
+}

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

@@ -0,0 +1,44 @@
+package mqtt
+
+import (
+	"context"
+	"fmt"
+
+	v2 "github.com/cloudevents/sdk-go/v2"
+	"github.com/eclipse/paho.golang/paho"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
+)
+
+func NewAgentOptions(opts *mqtt.MQTTOptions, clusterName, agentID string) *options.CloudEventsAgentOptions {
+	return &options.CloudEventsAgentOptions{
+		CloudEventsTransport: newTransport(
+			fmt.Sprintf("%s-client", agentID),
+			opts,
+			func(ctx context.Context, e v2.Event) (string, error) {
+				return mqtt.AgentPubTopic(ctx, opts, clusterName, e.Context)
+			},
+			func() (*paho.Subscribe, error) {
+				return mqtt.AgentSubscribe(opts, clusterName)
+			},
+		),
+		AgentID:     agentID,
+		ClusterName: clusterName,
+	}
+}
+
+func NewSourceOptions(opts *mqtt.MQTTOptions, clientID, sourceID string) *options.CloudEventsSourceOptions {
+	return &options.CloudEventsSourceOptions{
+		CloudEventsTransport: newTransport(
+			clientID,
+			opts,
+			func(ctx context.Context, e v2.Event) (string, error) {
+				return mqtt.SourcePubTopic(ctx, opts, sourceID, e.Context)
+			},
+			func() (*paho.Subscribe, error) {
+				return mqtt.SourceSubscribe(opts, sourceID)
+			},
+		),
+		SourceID: sourceID,
+	}
+}