Consume reaction source in SIMS

cmd/sims/consume.go

@@ -7,13 +7,15 @@ import (
 	"github.com/aws/aws-sdk-go/service/sqs"
 
 	"github.com/tapglue/snaas/core"
+	serr "github.com/tapglue/snaas/error"
 	"github.com/tapglue/snaas/platform/sns"
 	"github.com/tapglue/snaas/platform/source"
 	platformSQS "github.com/tapglue/snaas/platform/sqs"
 	"github.com/tapglue/snaas/service/app"
 	"github.com/tapglue/snaas/service/connection"
 	"github.com/tapglue/snaas/service/event"
 	"github.com/tapglue/snaas/service/object"
+	"github.com/tapglue/snaas/service/reaction"
 	"github.com/tapglue/snaas/service/rule"
 )
 
@@ -167,6 +169,50 @@ func consumeEvent(
 	}
 }
 
+func consumeReaction(
+	appFetch core.AppFetchFunc,
+	reactionSource reaction.Source,
+	batchc chan<- batch,
+	pipeline core.PipelineReactionFunc,
+	rules core.RuleListActiveFunc,
+) error {
+	for {
+		change, err := reactionSource.Consume()
+		if err != nil {
+			if serr.IsEmptySource(err) {
+				continue
+			}
+			return err
+		}
+
+		currentApp, err := appForNamespace(appFetch, change.Namespace)
+		if err != nil {
+			return err
+		}
+
+		rs, err := rules(currentApp, rule.TypeReaction)
+		if err != nil {
+			return err
+		}
+
+		ms, err := pipeline(currentApp, change, rs...)
+		if err != nil {
+			return err
+		}
+
+		if len(ms) == 0 {
+			err = reactionSource.Ack(change.AckID)
+			if err != nil {
+				return err
+			}
+
+			continue
+		}
+
+		batchc <- batchMessages(currentApp, reactionSource, change.AckID, ms)
+	}
+}
+
 func consumeObject(
 	appFetch core.AppFetchFunc,
 	objectSource object.Source,

cmd/sims/sims.go

@@ -6,7 +6,7 @@ import (
 	"os"
 	"time"
 
-	"github.com/tapglue/snaas/service/rule"
+	"github.com/tapglue/snaas/service/reaction"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
@@ -28,6 +28,7 @@ import (
 	"github.com/tapglue/snaas/service/event"
 	"github.com/tapglue/snaas/service/object"
 	"github.com/tapglue/snaas/service/platform"
+	"github.com/tapglue/snaas/service/rule"
 	"github.com/tapglue/snaas/service/user"
 )
 
@@ -258,6 +259,21 @@ func main() {
 	)(objectSource)
 	objectSource = object.LogSourceMiddleware(sourceService, logger)(objectSource)
 
+	reactionSource, err := reaction.SQSSource(sqsAPI)
+	if err != nil {
+		logger.Log("err", err, "lifecycle", "abort")
+		os.Exit(1)
+	}
+	reactionSource = reaction.InstrumentSourceMiddleware(
+		component,
+		sourceService,
+		sourceErrCount,
+		sourceOpCount,
+		sourceOpLatency,
+		sourceQueueLatency,
+	)(reactionSource)
+	reactionSource = reaction.LogSourceMiddleware(sourceService, logger)(reactionSource)
+
 	logger.Log(
 		"duration", time.Now().Sub(begin).Nanoseconds(),
 		"lifecycle", "start",
@@ -356,6 +372,20 @@ func main() {
 		}
 	}()
 
+	go func() {
+		err := consumeReaction(
+			core.AppFetch(apps),
+			reactionSource,
+			batchc,
+			core.PipelineReaction(objects, users),
+			core.RuleListActive(rules),
+		)
+		if err != nil {
+			logger.Log("err", err, "lifecycle", "abort")
+			os.Exit(1)
+		}
+	}()
+
 	// Distribute messages to channels.
 	cs := []channelFunc{
 		channelPush(

service/rule/rule.go

@@ -18,6 +18,7 @@ const (
 	TypeConnection Type = iota
 	TypeEvent
 	TypeObject
+	TypeReaction
 )
 
 type CriteriaConnection struct {