Adds request fifoMessageGroupId for SNS/SQS (#3415)

Signed-off-by: Josh Mangiola <contact@joshmangiola.com>

Author: joshm998

pubsub/aws/snssqs/snssqs.go

@@ -393,6 +393,9 @@ func (s *snsSqs) getOrCreateQueue(ctx context.Context, queueName string) (*sqsQu
 }
 
 func (s *snsSqs) getMessageGroupID(req *pubsub.PublishRequest) *string {
+	if reqGroupID := req.Metadata["fifoMessageGroupID"]; len(reqGroupID) > 0 {
+		return &reqGroupID
+	}
 	if len(s.metadata.FifoMessageGroupID) > 0 {
 		return &s.metadata.FifoMessageGroupID
 	}

pubsub/aws/snssqs/snssqs_test.go

@@ -508,3 +508,164 @@ func Test_buildARN_NonStandardPartition(t *testing.T) {
 	arn := ps.buildARN("sns", "myTopic")
 	r.Equal("arn:aws-cn:sns:cn-northwest-1:123456789012:myTopic", arn)
 }
+
+func Test_getMessageGroupID_RequestMessageGroupId(t *testing.T) {
+	t.Parallel()
+	r := require.New(t)
+	l := logger.NewLogger("SnsSqs unit test")
+	l.SetOutputLevel(logger.DebugLevel)
+	ps := snsSqs{
+		logger: l,
+		id:     "test-id",
+	}
+
+	md, err := ps.getSnsSqsMetadata(pubsub.Metadata{Base: metadata.Base{Properties: map[string]string{
+		"consumerID":         "c",
+		"accessKey":          "a",
+		"secretKey":          "s",
+		"fifo":               "true",
+		"fifoMessageGroupID": "test-group-id",
+	}}})
+	r.NoError(err)
+	ps.metadata = md
+
+	req := pubsub.PublishRequest{
+		Metadata:   map[string]string{"fifoMessageGroupID": "request-group-id"},
+		PubsubName: "test-pubsub",
+		Topic:      "test-topic",
+	}
+	groupID := ps.getMessageGroupID(&req)
+	r.Equal("request-group-id", *groupID)
+}
+
+func Test_getMessageGroupID_BlankRequestMessgeGroupId(t *testing.T) {
+	t.Parallel()
+	r := require.New(t)
+	l := logger.NewLogger("SnsSqs unit test")
+	l.SetOutputLevel(logger.DebugLevel)
+	ps := snsSqs{
+		logger: l,
+		id:     "test-id",
+	}
+
+	md, err := ps.getSnsSqsMetadata(pubsub.Metadata{Base: metadata.Base{Properties: map[string]string{
+		"consumerID":         "c",
+		"accessKey":          "a",
+		"secretKey":          "s",
+		"fifo":               "true",
+		"fifoMessageGroupID": "test-group-id",
+	}}})
+	r.NoError(err)
+	ps.metadata = md
+
+	req := pubsub.PublishRequest{
+		Metadata:   map[string]string{"fifoMessageGroupID": ""},
+		PubsubName: "test-pubsub",
+		Topic:      "test-topic",
+	}
+	groupID := ps.getMessageGroupID(&req)
+	r.Equal("test-group-id", *groupID)
+}
+
+func Test_getMessageGroupID_NoRequestMessageGroupId(t *testing.T) {
+	t.Parallel()
+	r := require.New(t)
+	l := logger.NewLogger("SnsSqs unit test")
+	l.SetOutputLevel(logger.DebugLevel)
+	ps := snsSqs{
+		logger: l,
+		id:     "test-id",
+	}
+
+	md, err := ps.getSnsSqsMetadata(pubsub.Metadata{Base: metadata.Base{Properties: map[string]string{
+		"consumerID":         "c",
+		"accessKey":          "a",
+		"secretKey":          "s",
+		"fifo":               "true",
+		"fifoMessageGroupID": "test-group-id",
+	}}})
+	r.NoError(err)
+	ps.metadata = md
+
+	req := pubsub.PublishRequest{
+		Metadata:   map[string]string{},
+		PubsubName: "test-pubsub",
+		Topic:      "test-topic",
+	}
+	groupID := ps.getMessageGroupID(&req)
+	r.Equal("test-group-id", *groupID)
+}
+
+func Test_getMessageGroupID(t *testing.T) {
+	t.Parallel()
+
+	fixtures := []struct {
+		name            string
+		componentProps  map[string]string
+		requestMetadata map[string]string
+		expectedGroupID string
+	}{
+		{
+			name:            "Request with fifoMessageGroupId",
+			componentProps:  map[string]string{"fifoMessageGroupID": "component-group-id"},
+			requestMetadata: map[string]string{"fifoMessageGroupID": "request-group-id"},
+			expectedGroupID: "request-group-id",
+		},
+		{
+			name:            "Request with blank fifoMessageGroupId",
+			componentProps:  map[string]string{"fifoMessageGroupID": "component-group-id"},
+			requestMetadata: map[string]string{"fifoMessageGroupID": ""},
+			expectedGroupID: "component-group-id",
+		},
+		{
+			name:            "Request without fifoMessageGroupId",
+			componentProps:  map[string]string{"fifoMessageGroupID": "component-group-id"},
+			requestMetadata: map[string]string{},
+			expectedGroupID: "component-group-id",
+		},
+		{
+			name:            "Component without fifoMessageGroupId",
+			componentProps:  map[string]string{},
+			requestMetadata: map[string]string{},
+			// Should be the default generated value
+			expectedGroupID: "c",
+		},
+	}
+
+	for _, f := range fixtures {
+		t.Run(f.name, func(t *testing.T) {
+			r := require.New(t)
+			l := logger.NewLogger("SnsSqs unit test")
+			l.SetOutputLevel(logger.DebugLevel)
+			ps := snsSqs{
+				logger: l,
+				id:     f.name,
+			}
+
+			props := map[string]string{
+				"consumerID": "c",
+				"accessKey":  "a",
+				"secretKey":  "s",
+				"fifo":       "true",
+			}
+			for k, v := range f.componentProps {
+				props[k] = v
+			}
+
+			md, err := ps.getSnsSqsMetadata(pubsub.Metadata{
+				Base: metadata.Base{Properties: props},
+			})
+
+			r.NoError(err)
+			ps.metadata = md
+
+			req := pubsub.PublishRequest{
+				Metadata:   f.requestMetadata,
+				PubsubName: "test-pubsub",
+				Topic:      "test-topic",
+			}
+			groupID := ps.getMessageGroupID(&req)
+			r.Equal(f.expectedGroupID, *groupID)
+		})
+	}
+}