separate the clientid and sourceid for mqtt source client

Signed-off-by: Wei Liu <[email protected]>
open-cluster-management-io · Dec 20, 2023 · 0738fe1 · 0738fe1
1 parent a2f58d6
commit 0738fe1
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 3 deletions.
diff --git a/cloudevents/generic/options/mqtt/sourceoptions.go b/cloudevents/generic/options/mqtt/sourceoptions.go
@@ -18,14 +18,16 @@ 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 {
 	return &options.CloudEventsSourceOptions{
 		CloudEventsOptions: &mqttSourceOptions{
 			MQTTOptions: *mqttOptions,
 			errorChan:   make(chan error),
 			sourceID:    sourceID,
+			clientID:    clientID,
 		},
 		SourceID: sourceID,
 	}
@@ -56,7 +58,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{},