From 51d4b5e5fcf5e85456ecf416edf86fcc207c308d Mon Sep 17 00:00:00 2001 From: Sergey Maskalik Date: Sat, 22 Feb 2020 12:47:49 -0800 Subject: [PATCH] Add support for FIFO queues (#7) * Add support for FIFO queues * Update readme for FIFO --- README.MD | 4 +++- main.go | 15 ++++++++++++++- 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/README.MD b/README.MD index 5c764f3..ee6a05a 100644 --- a/README.MD +++ b/README.MD @@ -8,8 +8,10 @@ to move deadletter queue messages back into the original queue. were successfully enqueued to the destination. * Messages are sent and received in batches for faster processing. * Progress indicator. +* Friendly output and error message. * Queue name resolution. For ease of use, you only need to provide a queue name and not the full `arn` address. -* Message Attributes are copied over. +* Message Attributes are copied over. +* Support for FIFO queues. MessageGroupId and MessageDeduplicationId are copied over to the destination messages. ## Installation diff --git a/main.go b/main.go index f963bee..808fb17 100644 --- a/main.go +++ b/main.go @@ -108,11 +108,21 @@ func logAwsError(message string, err error) { func convertToEntries(messages []*sqs.Message) []*sqs.SendMessageBatchRequestEntry { result := make([]*sqs.SendMessageBatchRequestEntry, len(messages)) for i, message := range messages { - result[i] = &sqs.SendMessageBatchRequestEntry{ + requestEntry := &sqs.SendMessageBatchRequestEntry{ MessageBody: message.Body, Id: message.MessageId, MessageAttributes: message.MessageAttributes, } + + if messageGroupId, ok := message.Attributes[sqs.MessageSystemAttributeNameMessageGroupId]; ok { + requestEntry.MessageGroupId = messageGroupId + } + + if messageDeduplicationId, ok := message.Attributes[sqs.MessageSystemAttributeNameMessageDeduplicationId]; ok { + requestEntry.MessageDeduplicationId = messageDeduplicationId + } + + result[i] = requestEntry } return result @@ -137,6 +147,9 @@ func moveMessages(sourceQueueUrl string, destinationQueueUrl string, svc *sqs.SQ WaitTimeSeconds: aws.Int64(0), MaxNumberOfMessages: aws.Int64(10), MessageAttributeNames: []*string{aws.String(sqs.QueueAttributeNameAll)}, + AttributeNames: []*string{ + aws.String(sqs.MessageSystemAttributeNameMessageGroupId), + aws.String(sqs.MessageSystemAttributeNameMessageDeduplicationId)}, } log.Info(color.New(color.FgCyan).Sprintf("Starting to move messages..."))