Merge branch 'master' into feature/maximum-message-size

Admiral-Piett · May 15, 2021 · 4792680 · 4792680
2 parents 46a9697 + 3993d6e
commit 4792680
Show file tree

Hide file tree

Showing 6 changed files with 289 additions and 4 deletions.
diff --git a/app/common.go b/app/common.go
@@ -36,6 +36,7 @@ type Environment struct {
 	AccountID              string
 	LogToFile              bool
 	LogFile                string
+	EnableDuplicates       bool
 	Topics                 []EnvTopic
 	Queues                 []EnvQueue
 	QueueAttributeDefaults EnvQueueAttributes

diff --git a/app/conf/config.go b/app/conf/config.go
@@ -5,6 +5,7 @@ import (
 	"io/ioutil"
 	"path/filepath"
 	"strings"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 
@@ -102,6 +103,8 @@ func LoadYamlConfig(filename string, env string) []string {
 			ReceiveWaitTimeSecs: queue.ReceiveMessageWaitTimeSeconds,
 			MaximumMessageSize:  queue.MaximumMessageSize,
 			IsFIFO:              app.HasFIFOQueueName(queue.Name),
+			EnableDuplicates:    app.CurrentEnvironment.EnableDuplicates,
+			Duplicates:          make(map[string]time.Time),
 		}
 	}
 
@@ -165,6 +168,8 @@ func createSqsSubscription(configSubscription app.EnvSubsciption, topicArn strin
 			ReceiveWaitTimeSecs: app.CurrentEnvironment.QueueAttributeDefaults.ReceiveMessageWaitTimeSeconds,
 			MaximumMessageSize:  app.CurrentEnvironment.QueueAttributeDefaults.MaximumMessageSize,
 			IsFIFO:              app.HasFIFOQueueName(configSubscription.QueueName),
+			EnableDuplicates:    app.CurrentEnvironment.EnableDuplicates,
+			Duplicates:          make(map[string]time.Time),
 		}
 	}
 	qArn := app.SyncQueues.Queues[configSubscription.QueueName].Arn

diff --git a/app/conf/goaws.yaml b/app/conf/goaws.yaml
@@ -10,6 +10,7 @@ Local:                              # Environment name that can be passed on the
   AccountId: "100010001000"
   LogToFile: false                 # Log messages (true/false)
   LogFile: .st/goaws_messages.log  # Log filename (for message logging
+  EnableDuplicates: false           # Enable or not deduplication based on messageDeduplicationId
   QueueAttributeDefaults:           # default attributes for all queues
     VisibilityTimeout: 30              # message visibility timeout
     ReceiveMessageWaitTimeSeconds: 0   # receive message max wait time

diff --git a/app/gosqs/gosqs.go b/app/gosqs/gosqs.go
@@ -56,6 +56,15 @@ func PeriodicTasks(d time.Duration, quit <-chan struct{}) {
 				log.Debugf("Queue [%s] length [%d]", queue.Name, len(queue.Messages))
 				for i := 0; i < len(queue.Messages); i++ {
 					msg := &queue.Messages[i]
+
+					// Reset deduplication period
+					for dedupId, startTime := range queue.Duplicates {
+						if time.Now().After(startTime.Add(app.DeduplicationPeriod)) {
+							log.Debugf("deduplication period for message with deduplicationId [%s] expired", dedupId)
+							delete(queue.Duplicates, dedupId)
+						}
+					}
+
 					if msg.ReceiptHandle != "" {
 						if msg.VisibilityTimeout.Before(time.Now()) {
 							log.Debugf("Making message visible again %s", msg.ReceiptHandle)
@@ -127,6 +136,8 @@ func CreateQueue(w http.ResponseWriter, req *http.Request) {
 			ReceiveWaitTimeSecs: app.CurrentEnvironment.QueueAttributeDefaults.ReceiveMessageWaitTimeSeconds,
 			MaximumMessageSize:  app.CurrentEnvironment.QueueAttributeDefaults.MaximumMessageSize,
 			IsFIFO:              app.HasFIFOQueueName(queueName),
+			EnableDuplicates:    app.CurrentEnvironment.EnableDuplicates,
+			Duplicates:          make(map[string]time.Time),
 		}
 		if err := validateAndSetQueueAttributes(queue, req.Form); err != nil {
 			createErrorResponse(w, req, err.Error())
@@ -149,6 +160,7 @@ func SendMessage(w http.ResponseWriter, req *http.Request) {
 	w.Header().Set("Content-Type", "application/xml")
 	messageBody := req.FormValue("MessageBody")
 	messageGroupID := req.FormValue("MessageGroupId")
+	messageDeduplicationID := req.FormValue("MessageDeduplicationId")
 	messageAttributes := extractMessageAttributes(req, "")
 
 	queueUrl := getQueueFromPath(req.FormValue("QueueUrl"), req.URL.String())
@@ -184,14 +196,22 @@ func SendMessage(w http.ResponseWriter, req *http.Request) {
 	msg.MD5OfMessageBody = common.GetMD5Hash(messageBody)
 	msg.Uuid, _ = common.NewUUID()
 	msg.GroupID = messageGroupID
+	msg.DeduplicationID = messageDeduplicationID
 	msg.SentTime = time.Now()
 
 	app.SyncQueues.Lock()
 	fifoSeqNumber := ""
 	if app.SyncQueues.Queues[queueName].IsFIFO {
 		fifoSeqNumber = app.SyncQueues.Queues[queueName].NextSequenceNumber(messageGroupID)
 	}
-	app.SyncQueues.Queues[queueName].Messages = append(app.SyncQueues.Queues[queueName].Messages, msg)
+
+	if !app.SyncQueues.Queues[queueName].IsDuplicate(messageDeduplicationID) {
+		app.SyncQueues.Queues[queueName].Messages = append(app.SyncQueues.Queues[queueName].Messages, msg)
+	} else {
+		log.Debugf("Message with deduplicationId [%s] in queue [%s] is duplicate ", messageDeduplicationID, queueName)
+	}
+
+	app.SyncQueues.Queues[queueName].InitDuplicatation(messageDeduplicationID)
 	app.SyncQueues.Unlock()
 	log.Infof("%s: Queue: %s, Message: %s\n", time.Now().Format("2006-01-02 15:04:05"), queueName, msg.MessageBody)
 
@@ -276,8 +296,12 @@ func SendMessageBatch(w http.ResponseWriter, req *http.Request) {
 				sendEntries[keyIndex-1].MessageGroupId = v[0]
 			}
 
-			if keySegments[2] == "MessageAttribute" {
-				sendEntries[keyIndex-1].MessageAttributes = extractMessageAttributes(req, strings.Join(keySegments[0:2], "."))
+			if keySegments[2] == "MessageGroupId" {
+				sendEntries[keyIndex-1].MessageGroupId = v[0]
+			}
+
+			if keySegments[2] == "MessageDeduplicationId" {
+				sendEntries[keyIndex-1].MessageDeduplicationId = v[0]
 			}
 		}
 	}
@@ -310,14 +334,23 @@ func SendMessageBatch(w http.ResponseWriter, req *http.Request) {
 		}
 		msg.MD5OfMessageBody = common.GetMD5Hash(sendEntry.MessageBody)
 		msg.GroupID = sendEntry.MessageGroupId
+		msg.DeduplicationID = sendEntry.MessageDeduplicationId
 		msg.Uuid, _ = common.NewUUID()
 		msg.SentTime = time.Now()
 		app.SyncQueues.Lock()
 		fifoSeqNumber := ""
 		if app.SyncQueues.Queues[queueName].IsFIFO {
 			fifoSeqNumber = app.SyncQueues.Queues[queueName].NextSequenceNumber(sendEntry.MessageGroupId)
 		}
-		app.SyncQueues.Queues[queueName].Messages = append(app.SyncQueues.Queues[queueName].Messages, msg)
+
+		if !app.SyncQueues.Queues[queueName].IsDuplicate(sendEntry.MessageDeduplicationId) {
+			app.SyncQueues.Queues[queueName].Messages = append(app.SyncQueues.Queues[queueName].Messages, msg)
+		} else {
+			log.Debugf("Message with deduplicationId [%s] in queue [%s] is duplicate ", sendEntry.MessageDeduplicationId, queueName)
+		}
+
+		app.SyncQueues.Queues[queueName].InitDuplicatation(sendEntry.MessageDeduplicationId)
+
 		app.SyncQueues.Unlock()
 		se := app.SendMessageBatchResultEntry{
 			Id:                     sendEntry.Id,
@@ -612,6 +645,7 @@ func DeleteMessageBatch(w http.ResponseWriter, req *http.Request) {
 					log.Printf("FIFO Queue %s unlocking group %s:", queueName, msg.GroupID)
 					app.SyncQueues.Queues[queueName].UnlockGroup(msg.GroupID)
 					app.SyncQueues.Queues[queueName].Messages = append(app.SyncQueues.Queues[queueName].Messages[:i], app.SyncQueues.Queues[queueName].Messages[i+1:]...)
+					delete(app.SyncQueues.Queues[queueName].Duplicates, msg.DeduplicationID)
 
 					deleteEntry.Deleted = true
 					deletedEntry := app.DeleteMessageBatchResultEntry{Id: deleteEntry.Id}
@@ -676,6 +710,7 @@ func DeleteMessage(w http.ResponseWriter, req *http.Request) {
 				app.SyncQueues.Queues[queueName].UnlockGroup(msg.GroupID)
 				//Delete message from Q
 				app.SyncQueues.Queues[queueName].Messages = append(app.SyncQueues.Queues[queueName].Messages[:i], app.SyncQueues.Queues[queueName].Messages[i+1:]...)
+				delete(app.SyncQueues.Queues[queueName].Duplicates, msg.DeduplicationID)
 
 				app.SyncQueues.Unlock()
 				// Create, encode/xml and send response
@@ -741,6 +776,7 @@ func PurgeQueue(w http.ResponseWriter, req *http.Request) {
 	app.SyncQueues.Lock()
 	if _, ok := app.SyncQueues.Queues[queueName]; ok {
 		app.SyncQueues.Queues[queueName].Messages = nil
+		app.SyncQueues.Queues[queueName].Duplicates = make(map[string]time.Time)
 		respStruct := app.PurgeQueueResponse{"http://queue.amazonaws.com/doc/2012-11-05/", app.ResponseMetadata{RequestId: "00000000-0000-0000-0000-000000000000"}}
 		enc := xml.NewEncoder(w)
 		enc.Indent("  ", "    ")