fix: Update Pipeline topics when writable pipeline changed

fixes edgexfoundry#1187

Goodell <[email protected]>

internal/app/configupdates.go

@@ -140,7 +140,9 @@ func (processor *ConfigUpdateProcessor) processConfigChangedPipeline() {
 
 		// Update the pipelines with their new transforms
 		for _, pipeline := range pipelines {
+			// TODO: Look at better way to apply pipeline updates
 			sdk.runtime.SetFunctionsPipelineTransforms(pipeline.Id, pipeline.Transforms)
+			sdk.runtime.SetFunctionsPipelineTopics(pipeline.Id, pipeline.Topics)
 		}
 
 		sdk.LoggingClient().Info("Configurable Pipeline successfully reloaded from new configuration")

internal/runtime/runtime.go

@@ -119,6 +119,20 @@ func (gr *GolangRuntime) SetFunctionsPipelineTransforms(id string, transforms []
 	}
 }
 
+// SetFunctionsPipelineTopics sets the topics for an existing function pipeline.
+// Non-existent pipelines are ignored
+func (gr *GolangRuntime) SetFunctionsPipelineTopics(id string, topics []string) {
+	pipeline := gr.pipelines[id]
+	if pipeline != nil {
+		gr.isBusyCopying.Lock()
+		pipeline.Topics = topics
+		gr.isBusyCopying.Unlock()
+		gr.lc.Infof("Topics set for `%s` pipeline", id)
+	} else {
+		gr.lc.Warnf("Unable to set topica for `%s` pipeline: Pipeline not found", id)
+	}
+}
+
 // ClearAllFunctionsPipelineTransforms clears the transforms for all existing function pipelines.
 func (gr *GolangRuntime) ClearAllFunctionsPipelineTransforms() {
 	gr.isBusyCopying.Lock()