separate the clientid and sourceid for mqtt source client (#305)

Signed-off-by: Wei Liu <[email protected]>
open-cluster-management-io · Dec 21, 2023 · 66fe32e · 66fe32e
1 parent 5aa27d6
commit 66fe32e
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 17 deletions.
diff --git a/cloudevents/generic/options/mqtt/agentoptions.go b/cloudevents/generic/options/mqtt/agentoptions.go
@@ -22,15 +22,17 @@ type mqttAgentOptions struct {
 }
 
 func NewAgentOptions(mqttOptions *MQTTOptions, clusterName, agentID string) *options.CloudEventsAgentOptions {
+	mqttAgentOptions := &mqttAgentOptions{
+		MQTTOptions: *mqttOptions,
+		errorChan:   make(chan error),
+		clusterName: clusterName,
+		agentID:     agentID,
+	}
+
 	return &options.CloudEventsAgentOptions{
-		CloudEventsOptions: &mqttAgentOptions{
-			MQTTOptions: *mqttOptions,
-			errorChan:   make(chan error),
-			clusterName: clusterName,
-			agentID:     agentID,
-		},
-		AgentID:     agentID,
-		ClusterName: clusterName,
+		CloudEventsOptions: mqttAgentOptions,
+		AgentID:            mqttAgentOptions.agentID,
+		ClusterName:        mqttAgentOptions.clusterName,
 	}
 }
 

diff --git a/cloudevents/generic/options/mqtt/sourceoptions.go b/cloudevents/generic/options/mqtt/sourceoptions.go
@@ -18,16 +18,20 @@ type mqttSourceOptions struct {
 	MQTTOptions
 	errorChan chan error
 	sourceID  string
+	clientID  string
 }
 
-func NewSourceOptions(mqttOptions *MQTTOptions, sourceID string) *options.CloudEventsSourceOptions {
+func NewSourceOptions(mqttOptions *MQTTOptions, clientID, sourceID string) *options.CloudEventsSourceOptions {
+	mqttSourceOptions := &mqttSourceOptions{
+		MQTTOptions: *mqttOptions,
+		errorChan:   make(chan error),
+		sourceID:    sourceID,
+		clientID:    clientID,
+	}
+
 	return &options.CloudEventsSourceOptions{
-		CloudEventsOptions: &mqttSourceOptions{
-			MQTTOptions: *mqttOptions,
-			errorChan:   make(chan error),
-			sourceID:    sourceID,
-		},
-		SourceID: sourceID,
+		CloudEventsOptions: mqttSourceOptions,
+		SourceID:           mqttSourceOptions.sourceID,
 	}
 }
 
@@ -56,7 +60,7 @@ func (o *mqttSourceOptions) WithContext(ctx context.Context, evtCtx cloudevents.
 func (o *mqttSourceOptions) Client(ctx context.Context) (cloudevents.Client, error) {
 	receiver, err := o.GetCloudEventsClient(
 		ctx,
-		fmt.Sprintf("%s-client", o.sourceID),
+		o.clientID,
 		func(err error) {
 			o.errorChan <- err
 		},

diff --git a/test/integration/cloudevents/source/client.go b/test/integration/cloudevents/source/client.go
@@ -111,9 +111,10 @@ func statusHashGetter(obj *Resource) (string, error) {
 }
 
 func StartMQTTResourceSourceClient(ctx context.Context, config *mqtt.MQTTOptions, eventHub *EventHub) (generic.CloudEventsClient[*Resource], error) {
+	sourceID := "integration-test"
 	client, err := generic.NewCloudEventSourceClient[*Resource](
 		ctx,
-		mqtt.NewSourceOptions(config, "integration-test"),
+		mqtt.NewSourceOptions(config, fmt.Sprintf("%s-client", sourceID), sourceID),
 		&resourceLister{},
 		statusHashGetter,
 		&resourceCodec{},