SD-103813 | implementation of the SQS receive message

inssqs/README.md

@@ -61,6 +61,9 @@ if err != nil {
 }
 ```
 
+### Receiving Messages
+Receive batch of messages from SQS queue using `ReceiveMessageBatch()`
+
 ## Configuration Options
 - Region: AWS region where the SQS queue resides.
 - QueueName: Name of the SQS queue.

inssqs/inssqs.go

@@ -2,6 +2,7 @@ package inssqs
 
 import (
 	"context"
+	"fmt"
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/aws/retry"
 	awsconfig "github.com/aws/aws-sdk-go-v2/config"
@@ -18,6 +19,7 @@ import (
 type Interface interface {
 	SendMessageBatch(entries []SQSMessageEntry) (failed []SQSMessageEntry, err error)
 	DeleteMessageBatch(entries []SQSDeleteMessageEntry) (failed []SQSDeleteMessageEntry, err error)
+	ReceiveMessageBatch(maxMessages, visibilityTimeout int32) (ReceiveMessageOutput, error)
 }
 
 type queue struct {
@@ -117,6 +119,39 @@ func (c *Config) setDefaults() {
 	}
 }
 
+// ReceiveMessageBatch receives a batch of messages from an SQS queue
+//
+// Parameters:
+// - maxMessages: An integer indicating the maximum number of messages to receive in a single batch.
+// - visibilityTimeout: An integer indicating the duration in seconds for which the received messages are hidden from subsequent retrieval.
+//
+// Returns:
+// - res: A pointer to ReceiveMessageOutput containing the received messages.
+// - err: An error indicating any failure during the receiving process, nil if all messages were received successfully.
+func (q *queue) ReceiveMessageBatch(maxMessages, visibilityTimeout int32) (ReceiveMessageOutput, error) {
+	if maxMessages > 10 {
+		return ReceiveMessageOutput{}, fmt.Errorf("maxMessages should be less than or equal to 10")
+	}
+
+	if maxMessages < 1 {
+		return ReceiveMessageOutput{}, fmt.Errorf("maxMessages should be greater than 0")
+	}
+
+	input := &awssqs.ReceiveMessageInput{
+		QueueUrl:            q.url,
+		MaxNumberOfMessages: maxMessages,
+		VisibilityTimeout:   visibilityTimeout,
+	}
+
+	res, err := q.client.ReceiveMessage(context.Background(), input)
+
+	if err != nil {
+		return ReceiveMessageOutput{}, err
+	}
+
+	return ReceiveMessageOutput{Messages: res.Messages}, nil
+}
+
 // SendMessageBatch sends a batch of messages to an SQS queue, handling retries and respecting batch size constraints.
 //
 // Parameters:

inssqs/model.go

@@ -1,6 +1,9 @@
 package inssqs
 
-import "github.com/aws/aws-sdk-go-v2/service/sqs/types"
+import (
+	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
+	"github.com/aws/smithy-go/middleware"
+)
 
 type entry interface {
 	getId() *string
@@ -43,3 +46,11 @@ func (e SQSDeleteMessageEntry) toDeleteMessageBatchRequestEntry() types.DeleteMe
 		ReceiptHandle: e.ReceiptHandle,
 	}
 }
+
+type ReceiveMessageOutput struct {
+	// A list of messages.
+	Messages []types.Message
+
+	// Metadata pertaining to the operation's result.
+	ResultMetadata middleware.Metadata
+}

inssqs/sqs/iface.go

@@ -9,6 +9,7 @@ type API interface {
 	SendMessageBatch(ctx context.Context, params *sqs.SendMessageBatchInput, optFns ...func(*sqs.Options)) (*sqs.SendMessageBatchOutput, error)
 	GetQueueUrl(ctx context.Context, params *sqs.GetQueueUrlInput, optFns ...func(*sqs.Options)) (*sqs.GetQueueUrlOutput, error)
 	DeleteMessageBatch(ctx context.Context, params *sqs.DeleteMessageBatchInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageBatchOutput, error)
+	ReceiveMessage(ctx context.Context, params *sqs.ReceiveMessageInput, optFns ...func(*sqs.Options)) (*sqs.ReceiveMessageOutput, error)
 }
 
 type proxy struct {
@@ -30,3 +31,7 @@ func (p *proxy) GetQueueUrl(ctx context.Context, params *sqs.GetQueueUrlInput, o
 func (p *proxy) DeleteMessageBatch(ctx context.Context, params *sqs.DeleteMessageBatchInput, optFns ...func(*sqs.Options)) (*sqs.DeleteMessageBatchOutput, error) {
 	return p.client.DeleteMessageBatch(ctx, params, optFns...)
 }
+
+func (p *proxy) ReceiveMessage(ctx context.Context, params *sqs.ReceiveMessageInput, optFns ...func(*sqs.Options)) (*sqs.ReceiveMessageOutput, error) {
+	return p.client.ReceiveMessage(ctx, params, optFns...)
+}

inssqs/sqs/sqs_mock.go

@@ -14,6 +14,26 @@ type MockAPI struct {
 	recorder *MockAPIMockRecorder
 }
 
+// ReceiveMessage mocks base method.
+func (m *MockAPI) ReceiveMessage(ctx context.Context, params *sqs.ReceiveMessageInput, optFns ...func(*sqs.Options)) (*sqs.ReceiveMessageOutput, error) {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{ctx, params}
+	for _, a := range optFns {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "ReceiveMessage", varargs...)
+	ret0, _ := ret[0].(*sqs.ReceiveMessageOutput)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ReceiveMessage indicates an expected call of ReceiveMessage.
+func (mr *MockAPIMockRecorder) ReceiveMessage(ctx, params interface{}, optFns ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{ctx, params}, optFns...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReceiveMessage", reflect.TypeOf((*MockAPI)(nil).ReceiveMessage), varargs...)
+}
+
 // MockAPIMockRecorder is the mock recorder for MockAPI.
 type MockAPIMockRecorder struct {
 	mock *MockAPI