refactor(triggers): Normalize Orchestration

Share functionality between custom and built-in triggers, expanding
testability of those built-in.

Signed-off-by: Alex Ullrich <[email protected]>

internal/app/service.go

@@ -160,7 +160,7 @@ func (svc *Service) MakeItRun() error {
 	svc.ctx.stop = stop
 
 	// determine input type and create trigger for it
-	t := svc.setupTrigger(svc.config, svc.runtime)
+	t := svc.setupTrigger(svc.config)
 	if t == nil {
 		return errors.New("failed to create Trigger")
 	}

internal/app/service_test.go

@@ -244,7 +244,7 @@ func TestSetupHTTPTrigger(t *testing.T) {
 	testRuntime := runtime.NewGolangRuntime("", nil, dic)
 	err := testRuntime.SetDefaultFunctionsPipeline(nil)
 	require.NoError(t, err)
-	trigger := sdk.setupTrigger(sdk.config, testRuntime)
+	trigger := sdk.setupTrigger(sdk.config)
 	result := IsInstanceOf(trigger, (*triggerHttp.Trigger)(nil))
 	assert.True(t, result, "Expected Instance of HTTP Trigger")
 }
@@ -261,7 +261,7 @@ func TestSetupMessageBusTrigger(t *testing.T) {
 	testRuntime := runtime.NewGolangRuntime("", nil, dic)
 	err := testRuntime.SetDefaultFunctionsPipeline(nil)
 	require.NoError(t, err)
-	trigger := sdk.setupTrigger(sdk.config, testRuntime)
+	trigger := sdk.setupTrigger(sdk.config)
 	result := IsInstanceOf(trigger, (*messagebus.Trigger)(nil))
 	assert.True(t, result, "Expected Instance of Message Bus Trigger")
 }

internal/app/customtrigger.go → internal/app/triggerfactory.go

@@ -20,21 +20,12 @@ package app
 
 import (
 	"fmt"
-	"github.com/edgexfoundry/go-mod-bootstrap/v2/di"
-	"github.com/edgexfoundry/go-mod-core-contracts/v2/clients/logger"
-	"github.com/hashicorp/go-multierror"
-	"strings"
-	"sync"
-
-	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
-
-	"github.com/edgexfoundry/app-functions-sdk-go/v2/internal/appfunction"
 	"github.com/edgexfoundry/app-functions-sdk-go/v2/internal/common"
-	"github.com/edgexfoundry/app-functions-sdk-go/v2/internal/runtime"
 	"github.com/edgexfoundry/app-functions-sdk-go/v2/internal/trigger/http"
 	"github.com/edgexfoundry/app-functions-sdk-go/v2/internal/trigger/messagebus"
 	"github.com/edgexfoundry/app-functions-sdk-go/v2/internal/trigger/mqtt"
 	"github.com/edgexfoundry/app-functions-sdk-go/v2/pkg/interfaces"
+	"strings"
 )
 
 const (
@@ -44,79 +35,25 @@ const (
 	TriggerTypeHTTP       = "HTTP"
 )
 
-type customTriggerBinding struct {
-	TriggerServiceBinding
-	dic *di.Container
-	log logger.LoggingClient
-}
-
-func newCustomTriggerBinding(svc *Service) *customTriggerBinding {
-	return &customTriggerBinding{
-		TriggerServiceBinding: simpleTriggerServiceBinding{
-			svc,
-			svc.runtime,
-		},
-		log: svc.LoggingClient(),
-		dic: svc.dic,
-	}
-}
-
-func (bnd *customTriggerBinding) buildContext(env types.MessageEnvelope) interfaces.AppFunctionContext {
-	return appfunction.NewContext(env.CorrelationID, bnd.dic, env.ContentType)
-}
-
-// processMessage provides an interface for custom triggers to execute service's configured pipelines through the runtime.
-func (bnd *customTriggerBinding) processMessage(ctx interfaces.AppFunctionContext, envelope types.MessageEnvelope) error {
-	bnd.log.Debugf("custom trigger attempting to find pipeline(s) for topic %s", envelope.ReceivedTopic)
-
-	// ensure we have a context established that we can safely cast to *appfunction.Context to pass to runtime
-	if _, ok := ctx.(*appfunction.Context); ctx == nil || !ok {
-		ctx = bnd.buildContext(envelope)
-	}
-
-	pipelines := bnd.GetMatchingPipelines(envelope.ReceivedTopic)
-
-	bnd.log.Debugf("custom trigger found %d pipeline(s) that match the incoming topic '%s'", len(pipelines), envelope.ReceivedTopic)
-
-	var finalErr error
-
-	pipelinesWaitGroup := sync.WaitGroup{}
-
-	for _, pipeline := range pipelines {
-		pipelinesWaitGroup.Add(1)
-		go func(p *interfaces.FunctionPipeline, wg *sync.WaitGroup, errCollector func(error)) {
-			defer wg.Done()
-
-			bnd.log.Tracef("custom trigger sending message to pipeline %s (%s)", p.Id, envelope.CorrelationID)
-
-			if msgErr := bnd.ProcessMessage(ctx.Clone().(*appfunction.Context), envelope, p); msgErr != nil {
-				bnd.log.Tracef("custom trigger message error in pipeline %s (%s) %s", p.Id, envelope.CorrelationID, msgErr.Err.Error())
-				errCollector(msgErr.Err)
-			}
-		}(pipeline, &pipelinesWaitGroup, func(e error) { finalErr = multierror.Append(finalErr, e) })
-	}
-
-	pipelinesWaitGroup.Wait()
-
-	return finalErr
-}
-
-func (svc *Service) setupTrigger(configuration *common.ConfigurationStruct, runtime *runtime.GolangRuntime) interfaces.Trigger {
+func (svc *Service) setupTrigger(configuration *common.ConfigurationStruct) interfaces.Trigger {
 	var t interfaces.Trigger
-	// Need to make dynamic, search for the trigger that is input
+
+	// this is an extension on the standard trigger binding and can be used for both
+	bnd := NewTriggerServiceBinding(svc)
+	mp := &triggerMessageProcessor{bnd}
 
 	switch triggerType := strings.ToUpper(configuration.Trigger.Type); triggerType {
 	case TriggerTypeHTTP:
 		svc.LoggingClient().Info("HTTP trigger selected")
-		t = http.NewTrigger(svc.dic, runtime, svc.webserver)
+		t = http.NewTrigger(bnd, mp, svc.webserver)
 
 	case TriggerTypeMessageBus:
 		svc.LoggingClient().Info("EdgeX MessageBus trigger selected")
-		t = messagebus.NewTrigger(svc.dic, runtime)
+		t = messagebus.NewTrigger(bnd, mp)
 
 	case TriggerTypeMQTT:
 		svc.LoggingClient().Info("External MQTT trigger selected")
-		t = mqtt.NewTrigger(svc.dic, runtime)
+		t = mqtt.NewTrigger(bnd, mp)
 
 	default:
 		if factory, found := svc.customTriggerFactories[triggerType]; found {
@@ -150,12 +87,13 @@ func (svc *Service) RegisterCustomTriggerFactory(name string,
 	}
 
 	svc.customTriggerFactories[nu] = func(sdk *Service) (interfaces.Trigger, error) {
-		binding := newCustomTriggerBinding(sdk)
+		binding := NewTriggerServiceBinding(sdk)
+		messageProcessor := &triggerMessageProcessor{binding}
 
 		cfg := interfaces.TriggerConfig{
 			Logger:           sdk.lc,
-			ContextBuilder:   binding.buildContext,
-			MessageProcessor: binding.processMessage,
+			ContextBuilder:   binding.BuildContext,
+			MessageProcessor: messageProcessor.Process,
 			ConfigLoader:     binding.LoadCustomConfig,
 		}