Skip to content

Commit

Permalink
Adds the ability to use SQS FIFO queues.
Browse files Browse the repository at this point in the history
All SQS FIFO queues have the `.fifo` suffix at the end of their URL - this is how we determine if we are using a FIFO queue or not. The `MessageGroupId` (A required field for FIFO) is set using the key & id of the event's message.

Note: This commit does not add the the `MessageDeduplicationID` so the FIFO queue must have content-based deduplication enabled.
  • Loading branch information
crankycookie committed Dec 5, 2024
1 parent d445566 commit d24fbd0
Showing 1 changed file with 13 additions and 1 deletion.
14 changes: 13 additions & 1 deletion internal/endpoint/sqs.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@ import (
"sync"
"time"

"github.com/tidwall/gjson"

"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/credentials"
"github.com/aws/aws-sdk-go/aws/session"
Expand Down Expand Up @@ -108,11 +110,17 @@ func (conn *SQSConn) Send(msg string) error {
}

queueURL := conn.generateSQSURL()
// Send message
// Create message
sendParams := &sqs.SendMessageInput{
MessageBody: aws.String(msg),
QueueUrl: aws.String(queueURL),
}
if isFifoQueue(queueURL) {
key := gjson.Get(msg, "key")
id := gjson.Get(msg, "id")
keyValue := fmt.Sprintf("%s#%s", key.String(), id.String())
sendParams.MessageGroupId = aws.String(keyValue)
}
_, err := conn.svc.SendMessage(sendParams)
if err != nil {
fmt.Println(err)
Expand Down Expand Up @@ -145,3 +153,7 @@ func sqsRegionFromPlainURL(s string) string {
}
return ""
}

func isFifoQueue(s string) bool {
return strings.HasSuffix(s, ".fifo")
}

0 comments on commit d24fbd0

Please sign in to comment.