mqtt transport v2

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

Author: skeeey

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

@@ -38,7 +38,7 @@ func newTransport(clientID string, opts *mqtt.MQTTOptions, pubTopicGetter pubTop
 	return &mqttTransport{
 		opts:            opts,
 		clientID:        clientID,
-		errorChan:       make(chan error),
+		errorChan:       make(chan error, 1),
 		getPublishTopic: pubTopicGetter,
 		getSubscribe:    subscribeGetter,
 	}

test/integration/cloudevents/certrotation_mqtt_test.go

@@ -10,6 +10,7 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/cert"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
+	mqttv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/mqtt"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 	"open-cluster-management.io/sdk-go/test/integration/cloudevents/util"
 )
@@ -18,7 +19,7 @@ var _ = ginkgo.Describe("CloudEvents Certificate Rotation Test - MQTT", runCloud
 
 func GetMQTTAgentOptions(_ context.Context, agentID, clusterName, clientCertFile, clientKeyFile string) *options.CloudEventsAgentOptions {
 	mqttOptions := newTLSMQTTOptions(certPool, mqttTLSBrokerHost, clientCertFile, clientKeyFile)
-	return mqtt.NewAgentOptions(mqttOptions, clusterName, agentID)
+	return mqttv2.NewAgentOptions(mqttOptions, clusterName, agentID)
 }
 
 func newTLSMQTTOptions(certPool *x509.CertPool, brokerHost, clientCertFile, clientKeyFile string) *mqtt.MQTTOptions {

test/integration/cloudevents/cloudevents_resync_test.go

@@ -4,9 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"time"
+
 	jsonpatch "github.com/evanphx/json-patch/v5"
 	"k8s.io/apimachinery/pkg/types"
-	"time"
 
 	"github.com/onsi/ginkgo"
 	"github.com/onsi/gomega"
@@ -20,6 +21,7 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/agent/codec"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
+	mqttv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/mqtt"
 	"open-cluster-management.io/sdk-go/test/integration/cloudevents/agent"
 	"open-cluster-management.io/sdk-go/test/integration/cloudevents/source"
 	"open-cluster-management.io/sdk-go/test/integration/cloudevents/store"
@@ -55,7 +57,7 @@ var _ = ginkgo.Describe("CloudEvents Clients Test - RESYNC", func() {
 
 		sourceCloudEventsClient, err = source.StartResourceSourceClient(
 			ctx,
-			mqtt.NewSourceOptions(
+			mqttv2.NewSourceOptions(
 				util.NewMQTTSourceOptions(mqttBrokerHost, sourceID),
 				fmt.Sprintf("%s-client", sourceID),
 				sourceID,

test/integration/cloudevents/cloudevetns_mqtt_test.go

@@ -8,14 +8,14 @@ import (
 
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/constants"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
+	mqttv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/mqtt"
 	"open-cluster-management.io/sdk-go/test/integration/cloudevents/util"
 )
 
 var _ = ginkgo.Describe("CloudEvents Clients Test - MQTT", runCloudeventsClientPubSubTest(GetMQTTSourceOptions))
 
 func GetMQTTSourceOptions(_ context.Context, sourceID string) (*options.CloudEventsSourceOptions, string) {
-	return mqtt.NewSourceOptions(
+	return mqttv2.NewSourceOptions(
 		util.NewMQTTSourceOptions(mqttBrokerHost, sourceID),
 		fmt.Sprintf("%s-client", sourceID),
 		sourceID,