Repository: beatlabs/patron
Pull request #520 (2022-06-17 09:03:19)

Add AWS Queue Owner ID
<!--
Thanks for taking precious time for making a PR.

Before creating a pull request, please make sure:
- Your PR solves one problem for which an issue exist and a solution has been discussed
- You have read the guide for contributing
  - See https://github.com/beatlabs/patron/blob/master/README.md#how-to-contribute
- You signed all your commits (otherwise we won't be able to merge the PR)
  - See https://github.com/beatlabs/patron/blob/master/SIGNYOURWORK.md
- You added unit tests for the new functionality
- You mention in the PR description which issue it is addressing, e.g. "Resolves #123"
-->

## Which problem is this PR solving?

Solves issue [#518  Add AWS Queue Owner Account ID for SQS](https://github.com/beatlabs/patron/issues/518)

## Short description of the changes

Added a new option function in order to be able to pass the Queue Owner Account ID. 
If a valid ID exists then it is passed in to the `GetQueueUrlWithContext` call.


erifili117: Hey @mantzas! I can see the `Waiting for status to be reported`, in CI, for about 40 minutes now. Did the job get stuck? Thanks!  

Patch:
diff --git a/component/sqs/component.go b/component/sqs/component.go
index efc2baa86..1b62f3cd5 100644
--- a/component/sqs/component.go
+++ b/component/sqs/component.go
@@ -101,13 +101,14 @@ type stats struct {
 
 // Component implementation of an async component.
 type Component struct {
-	name  string
-	queue queue
-	api   sqsiface.SQSAPI
-	cfg   config
-	proc  ProcessorFunc
-	stats stats
-	retry retry
+	name       string
+	queue      queue
+	queueOwner string
+	api        sqsiface.SQSAPI
+	cfg        config
+	proc       ProcessorFunc
+	stats      stats
+	retry      retry
 }
 
 // New creates a new component with support for functional configuration.
@@ -128,18 +129,10 @@ func New(name, queueName string, sqsAPI sqsiface.SQSAPI, proc ProcessorFunc, oo
 		return nil, errors.New("process function is nil")
 	}
 
-	out, err := sqsAPI.GetQueueUrlWithContext(context.Background(), &sqs.GetQueueUrlInput{
-		QueueName: aws.String(queueName),
-	})
-	if err != nil {
-		return nil, fmt.Errorf("failed to get queue URL: %w", err)
-	}
-
 	cmp := &Component{
 		name: name,
 		queue: queue{
 			name: queueName,
-			url:  aws.StringValue(out.QueueUrl),
 		},
 		api: sqsAPI,
 		cfg: config{
@@ -156,12 +149,27 @@ func New(name, queueName string, sqsAPI sqsiface.SQSAPI, proc ProcessorFunc, oo
 	}
 
 	for _, optionFunc := range oo {
-		err = optionFunc(cmp)
+		err := optionFunc(cmp)
 		if err != nil {
 			return nil, err
 		}
 	}
 
+	req := &sqs.GetQueueUrlInput{
+		QueueName: aws.String(queueName),
+	}
+
+	if cmp.queueOwner != "" {
+		req.SetQueueOwnerAWSAccountId(cmp.queueOwner)
+	}
+
+	out, err := sqsAPI.GetQueueUrlWithContext(context.Background(), req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get queue URL: %w", err)
+	}
+
+	cmp.queue.url = aws.StringValue(out.QueueUrl)
+
 	return cmp, nil
 }
 
diff --git a/component/sqs/option.go b/component/sqs/option.go
index 7a42f4c43..8b7e6aca4 100644
--- a/component/sqs/option.go
+++ b/component/sqs/option.go
@@ -80,3 +80,14 @@ func RetryWait(interval time.Duration) OptionFunc {
 		return nil
 	}
 }
+
+// QueueOwner sets the AWS queue owner ID for an SQS queue.
+func QueueOwner(queueOwner string) OptionFunc {
+	return func(c *Component) error {
+		if queueOwner == "" {
+			return errors.New("queue owner should not be empty")
+		}
+		c.queueOwner = queueOwner
+		return nil
+	}
+}


Test patch:
diff --git a/component/sqs/component_test.go b/component/sqs/component_test.go
index 5f83babcd..178ab5f88 100644
--- a/component/sqs/component_test.go
+++ b/component/sqs/component_test.go
@@ -98,6 +98,25 @@ func TestNew(t *testing.T) {
 			},
 			expectedErr: "retry wait time should be a positive number",
 		},
+		"success queue owner": {
+			args: args{
+				name:      "name",
+				queueName: "queueName",
+				sqsAPI:    &stubSQSAPI{},
+				proc:      sp.process,
+				oo:        []OptionFunc{QueueOwner("10201020")},
+			},
+		},
+		"queue owner fails": {
+			args: args{
+				name:      "name",
+				queueName: "queueName",
+				sqsAPI:    &stubSQSAPI{},
+				proc:      sp.process,
+				oo:        []OptionFunc{QueueOwner("")},
+			},
+			expectedErr: "queue owner should not be empty",
+		},
 	}
 	for name, tt := range tests {
 		tt := tt
diff --git a/component/sqs/option_test.go b/component/sqs/option_test.go
index 50da784d5..16efa6a58 100644
--- a/component/sqs/option_test.go
+++ b/component/sqs/option_test.go
@@ -191,3 +191,36 @@ func TestRetryWait(t *testing.T) {
 		})
 	}
 }
+
+func TestQueueOwner(t *testing.T) {
+	t.Parallel()
+	type args struct {
+		queueOwner string
+	}
+	tests := map[string]struct {
+		args        args
+		expectedErr string
+	}{
+		"success": {
+			args: args{queueOwner: "10201020"},
+		},
+		"empty queue owner": {
+			args:        args{queueOwner: ""},
+			expectedErr: "queue owner should not be empty",
+		},
+	}
+	for name, tt := range tests {
+		tt := tt
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			c := &Component{}
+			err := QueueOwner(tt.args.queueOwner)(c)
+			if tt.expectedErr != "" {
+				assert.EqualError(t, err, tt.expectedErr)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, c.queueOwner, tt.args.queueOwner)
+			}
+		})
+	}
+}