Merge pull request #175 from jackchenjc/bootstrap-issue-468

feat!: Passing callback function into WatchForChanges method

configuration/interface.go

@@ -1,6 +1,6 @@
 //
 // Copyright (c) 2019 Intel Corporation
-// Copyright (C) 2024 IOTech Ltd
+// Copyright (C) 2024-2025 IOTech Ltd
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -41,7 +41,7 @@ type Client interface {
 	// WatchForChanges sets up a keeper watch for the target key and send back updates on the update channel.
 	// Passed in struct is only a reference for Configuration service, empty struct is ok
 	// Sends the configuration in the target struct as interface{} on updateChannel, which caller must cast
-	WatchForChanges(updateChannel chan<- interface{}, errorChannel chan<- error, configuration interface{}, waitKey string, msgClient messaging.MessageClient)
+	WatchForChanges(updateChannel chan<- interface{}, errorChannel chan<- error, configuration interface{}, waitKey string, getMsgClientCb func() messaging.MessageClient)
 
 	// StopWatching causes all WatchForChanges processing to stop and waits until they have stopped.
 	StopWatching()

internal/pkg/keeper/client.go

@@ -1,5 +1,5 @@
 //
-// Copyright (C) 2024 IOTech Ltd
+// Copyright (C) 2024-2025 IOTech Ltd
 //
 // SPDX-License-Identifier: Apache-2.0
 
@@ -167,8 +167,9 @@ func (k *keeperClient) GetConfiguration(configStruct interface{}) (interface{},
 	return configStruct, nil
 }
 
-func (k *keeperClient) WatchForChanges(updateChannel chan<- interface{}, errorChannel chan<- error, configuration interface{}, waitKey string, messageBus messaging.MessageClient) {
-	if messageBus == nil {
+func (k *keeperClient) WatchForChanges(updateChannel chan<- interface{}, errorChannel chan<- error, configuration interface{}, waitKey string, getMsgClientCb func() messaging.MessageClient) {
+	messageClient := getMsgClientCb()
+	if messageClient == nil {
 		configErr := errors.New("unable to use MessageClient to watch for configuration changes")
 		errorChannel <- configErr
 		return
@@ -184,16 +185,16 @@ func (k *keeperClient) WatchForChanges(updateChannel chan<- interface{}, errorCh
 	}
 
 	watchErrors := make(chan error)
-	err := messageBus.Subscribe(topics, watchErrors)
+	err := messageClient.Subscribe(topics, watchErrors)
 	if err != nil {
-		_ = messageBus.Disconnect()
+		_ = messageClient.Disconnect()
 		errorChannel <- err
 		return
 	}
 
 	go func() {
 		defer func() {
-			_ = messageBus.Disconnect()
+			_ = messageClient.Disconnect()
 		}()
 
 		// send a nil value to updateChannel once the watcher connection is established