feat(sdk): Support Multi-Topic Pipeline Configuration #947

AlexCuse · 2021-09-13T21:08:22Z

Match incoming topic against an array of topics when determining whether
or not to execute a given pipeline. Allow Topics configuration as comma
separated list similar to messaging triggers.

PR Checklist

Please check if your PR fulfills the following requirements:

Tests for the changes have been added (for bug fixes / features)
Docs have been added / updated (for bug fixes / features)

Related Docs PR now required (if applicable)

Related Docs PR:

edgexfoundry/edgex-docs#551

If n/a for Docs PR, state why it is not applicable:

What is the current behavior?

Pipelines can only be configured to run for a single topic.

Issue Number: #945

What is the new behavior?

Pipelines can be configured with a comma-separated list of topics similar to messaging triggers.

Does this PR introduce a breaking change?

Yes
No

Are there any new imports or modules? If so, what are they used for and why?

Yes
No

Are there any specific instructions or things that should be known prior to reviewing?

Other information

codecov-commenter · 2021-09-13T21:18:23Z

Codecov Report

Merging #947 (9314ea9) into main (29a7969) will increase coverage by 0.08%.
The diff coverage is 94.59%.

@@            Coverage Diff             @@
##             main     #947      +/-   ##
==========================================
+ Coverage   65.63%   65.71%   +0.08%     
==========================================
  Files          33       33              
  Lines        2514     2520       +6     
==========================================
+ Hits         1650     1656       +6     
  Misses        764      764              
  Partials      100      100

Impacted Files	Coverage Δ
internal/runtime/runtime.go	`76.74% <92.59%> (+0.32%)`	⬆️
internal/app/service.go	`55.66% <100.00%> (+0.42%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 29a7969...9314ea9. Read the comment docs.

lenny-goodell · 2021-09-13T21:14:58Z

app-service-template/main.go

@@ -100,7 +100,7 @@ func (app *myApp) CreateAndRunAppService(serviceKey string, newServiceFactory fu
 	sample := functions.NewSample()

 	// TODO: Replace below functions with built in and/or your custom functions for your use case
-	//       or remove is using Pipeline By Topic below.
+	//       or remove is using Pipeline By Topics below.


Suggested change

// or remove is using Pipeline By Topics below.

// or remove if using Pipeline By Topics below.

lenny-goodell · 2021-09-13T21:18:47Z

app-service-template/main.go

@@ -120,8 +120,8 @@ func (app *myApp) CreateAndRunAppService(serviceKey string, newServiceFactory fu
 	//       Core Data publishes to the 'edgex/events/core/<profile-name>/<device-name>/<source-name>' topic
 	// Note: This example with default above causes Events from Random-Float-Device device to be processed twice
 	//       resulting in the XML to be published back to the MessageBus twice.
-	// See <Pipeline By Topic documentation url TBD> for more details.
-	err = app.service.AddFunctionsPipelineForTopic("Floats", "edgex/events/#/#/Random-Float-Device/#",
+	// See <Pipeline By Topics documentation url TBD> for more details.


Suggested change

// See <Pipeline By Topics documentation url TBD> for more details.

// See https://docs.edgexfoundry.org/2.0/microservices/application/AdvancedTopics/#pipeline-per-topics for more details.

lenny-goodell · 2021-09-13T21:20:58Z

internal/app/service.go

-	if len(strings.TrimSpace(topic)) == 0 {
-		return errors.New("topic for pipeline can not be blank")
+	if len(topics) == 0 {
+		return errors.New("topics for pipeline can not be blank")


Suggested change

return errors.New("topics for pipeline can not be blank")

return errors.New("topics for pipeline can not be empty")

lenny-goodell · 2021-09-13T21:26:41Z

internal/app/service_test.go

-		{"Duplicate Id", "123", TriggerTypeMessageBus, "#", transforms, true},
-		{"HTTP Trigger Type", "127", TriggerTypeHTTP, "#", transforms, true},
+		{"Happy Path", "123", TriggerTypeMessageBus, defaultTopics, transforms, false},
+		{"Duplicate Id", "123", TriggerTypeMessageBus, defaultTopics, transforms, true},


This test depends on pervious test to have run. Correct? This is not good practice as I can't run this test by itself. This is way I used service.SetDefaultFunctionsPipeline above. Does look like I had something else messaged up... ;-) "Default Id", interfaces.DefaultPipelineId, should have been the "Duplicate Id"

lenny-goodell · 2021-09-13T21:28:17Z

internal/common/config.go

@@ -162,12 +162,12 @@ type PipelineInfo struct {
 	Functions map[string]PipelineFunction
 }

-// TopicPipeline define the data to a Pre Topic functions pipeline
+// TopicPipeline define the data to a Pre Topics functions pipeline


Suggested change

// TopicPipeline define the data to a Pre Topics functions pipeline

// TopicPipeline define the data to a Per Topics functions pipeline

lenny-goodell · 2021-09-13T21:33:26Z

internal/runtime/runtime.go

+		wildcardCount := strings.Count(pipelineTopic, TopicWildCard)
+		switch wildcardCount {
+		case 0:
+			return incomingTopic == pipelineTopic


Suggested change

return incomingTopic == pipelineTopic

if incomingTopic == pipelineTopic {

return true

}

lenny-goodell · 2021-09-13T21:34:25Z

internal/runtime/runtime.go

-			return false
-		}
+			if len(pipelineLevels) > len(incomingLevels) {
+				return false


Suggested change

return false

continue

lenny-goodell · 2021-09-13T21:45:42Z

internal/runtime/runtime_test.go

@@ -566,7 +566,7 @@ func TestTopicMatches(t *testing.T) {

 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			actual := topicMatches(test.incomingTopic, test.pipelineTopic)
+			actual := topicMatches(test.incomingTopic, []string{test.pipelineTopic})


Expand the tests struct to have multiple topics and add cases for:

first topic has no wild card and doesn't match exactly and second topic does match in some way

first topic has levels greater than that of incoming and second topic does match in some way

Also looks like I was missing a test for a single pipeline topic that has levels greater than that of incoming

lenny-goodell · 2021-09-13T21:53:09Z

pkg/interfaces/service.go

@@ -88,11 +88,11 @@ type ApplicationService interface {
 	// Note that the functions are executed in the order provided in the list.
 	// An error is returned if the list is empty.
 	SetDefaultFunctionsPipeline(transforms ...AppFunction) error
-	// AddFunctionsPipelineForTopic adds a functions pipeline with the specified unique id and list of Application Functions
+	// AddFunctionsPipelineForTopics adds a functions pipeline with the specified unique id and list of Application Functions
 	// to be executed when the incoming topic matches the specified topic. The specified topic may contain the '#' wildcard


Suggested change

// to be executed when the incoming topic matches the specified topic. The specified topic may contain the '#' wildcard

// to be executed when the incoming topic matches any of the specified topics. The specified topic may contain the '#' wildcard

lenny-goodell · 2021-09-13T21:58:27Z

internal/common/config.go

 type TopicPipeline struct {
 	// Id is the unique ID of the pipeline instance
 	Id string
-	// Topic is the topic which must match the incoming topic inorder for the pipeline to execute
-	Topic string
+	// Topics is the set of topics matched against the incoming to determine if pipeline should execute


Suggested change

// Topics is the set of topics matched against the incoming to determine if pipeline should execute

// Topics is the set of comma separated topics matched against the incoming to determine if pipeline should execute

Match incoming topic against an array of topics when determining whether or not to execute a given pipeline. Allow Topics configuration as comma separated list similar to messaging triggers. Also allow configuration of non-default pipeline(s) for HTTP trigger. Signed-off-by: Alex Ullrich <[email protected]>

lenny-goodell

LGTM

Match incoming topic against an array of topics when determining whether or not to execute a given pipeline. Allow Topics configuration as comma separated list similar to messaging triggers. Also allow configuration of non-default pipeline(s) for HTTP trigger. Signed-off-by: Alex Ullrich <[email protected]>

lenny-goodell requested changes Sep 13, 2021

View reviewed changes

AlexCuse force-pushed the main branch from dea0737 to 9314ea9 Compare September 13, 2021 23:37

AlexCuse force-pushed the main branch from 9314ea9 to 3030fa0 Compare September 14, 2021 18:49

lenny-goodell approved these changes Sep 14, 2021

View reviewed changes

lenny-goodell merged commit 0793683 into edgexfoundry:main Sep 14, 2021

lenny-goodell linked an issue Sep 15, 2021 that may be closed by this pull request

[SDK] Allow Topic-Specific Pipelines to Be Configured with Multiple Topics #945

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(sdk): Support Multi-Topic Pipeline Configuration #947

feat(sdk): Support Multi-Topic Pipeline Configuration #947

AlexCuse commented Sep 13, 2021

codecov-commenter commented Sep 13, 2021 •

edited

Loading

lenny-goodell Sep 13, 2021

lenny-goodell Sep 13, 2021

lenny-goodell Sep 13, 2021

lenny-goodell Sep 13, 2021 •

edited

Loading

lenny-goodell Sep 13, 2021

lenny-goodell Sep 13, 2021

lenny-goodell Sep 13, 2021

lenny-goodell Sep 13, 2021

lenny-goodell Sep 13, 2021

lenny-goodell Sep 13, 2021

lenny-goodell left a comment

	// or remove is using Pipeline By Topics below.
	// or remove if using Pipeline By Topics below.

	// See <Pipeline By Topics documentation url TBD> for more details.
	// See https://docs.edgexfoundry.org/2.0/microservices/application/AdvancedTopics/#pipeline-per-topics for more details.

	return errors.New("topics for pipeline can not be blank")
	return errors.New("topics for pipeline can not be empty")

	// TopicPipeline define the data to a Pre Topics functions pipeline
	// TopicPipeline define the data to a Per Topics functions pipeline

-			return incomingTopic == pipelineTopic
+			if incomingTopic == pipelineTopic {
+				return true
+			}

	// to be executed when the incoming topic matches the specified topic. The specified topic may contain the '#' wildcard
	// to be executed when the incoming topic matches any of the specified topics. The specified topic may contain the '#' wildcard

	// Topics is the set of topics matched against the incoming to determine if pipeline should execute
	// Topics is the set of comma separated topics matched against the incoming to determine if pipeline should execute

feat(sdk): Support Multi-Topic Pipeline Configuration #947

feat(sdk): Support Multi-Topic Pipeline Configuration #947

Conversation

AlexCuse commented Sep 13, 2021

PR Checklist

Related Docs PR now required (if applicable)

What is the current behavior?

What is the new behavior?

Does this PR introduce a breaking change?

Are there any new imports or modules? If so, what are they used for and why?

Are there any specific instructions or things that should be known prior to reviewing?

Other information

codecov-commenter commented Sep 13, 2021 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lenny-goodell Sep 13, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lenny-goodell left a comment

Choose a reason for hiding this comment

codecov-commenter commented Sep 13, 2021 •

edited

Loading

lenny-goodell Sep 13, 2021 •

edited

Loading