Repository: softwaremill/elasticmq
Pull request #527 (2021-09-08 10:51:36)

Add sequence number to FIFO queue
According to [Message docs](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_Message.html), there should be additional attribute in `ReceiveMessage` - `SequenceNumber`. Other than that `SequenceNumber` should be returned in `SendMessage` result as per [this doc](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html).

Number should be increasing per GroupId.

Few things I've noticed on AWS:
- SequenceNumber is increasing globally, but in docs it is specified that it should increase at least per GroupId.
- SequenceNumber passed as an attribute in endMessage is ignored when receiving the same message

Connected with issue #522 


Patch:
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/InternalMessage.scala b/core/src/main/scala/org/elasticmq/actor/queue/InternalMessage.scala
index 9cc343d8..e7257235 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/InternalMessage.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/InternalMessage.scala
@@ -83,8 +83,8 @@ case class InternalMessage(
 }
 
 object InternalMessage {
-
   def from(newMessageData: NewMessageData, queueData: QueueData): InternalMessage = {
+
     val now = System.currentTimeMillis()
     new InternalMessage(
       newMessageData.id.getOrElse(generateId()).id,
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/QueueActorStorage.scala b/core/src/main/scala/org/elasticmq/actor/queue/QueueActorStorage.scala
index e973930c..de56d845 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/QueueActorStorage.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/QueueActorStorage.scala
@@ -13,8 +13,15 @@ trait QueueActorStorage {
   def copyMessagesToActorRef: Option[ActorRef]
   def moveMessagesToActorRef: Option[ActorRef]
 
+  def nextSequenceNumber(): Long = {
+    val next = sequenceNumber
+    sequenceNumber = sequenceNumber + 1
+    next
+  }
+
   var queueData: QueueData = initialQueueData
   var messageQueue: MessageQueue = MessageQueue(queueData.isFifo)
   var fifoMessagesHistory: FifoDeduplicationIdsHistory = FifoDeduplicationIdsHistory.newHistory()
   val receiveRequestAttemptCache = new ReceiveRequestAttemptCache
+  private var sequenceNumber: Long = 0
 }
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/operations/SendMessageOp.scala b/core/src/main/scala/org/elasticmq/actor/queue/operations/SendMessageOp.scala
index 9db4dc14..3299f374 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/operations/SendMessageOp.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/operations/SendMessageOp.scala
@@ -5,7 +5,7 @@ import org.elasticmq.actor.queue.{InternalMessage, QueueActorStorage}
 import org.elasticmq.actor.reply.{DoNotReply, ReplyAction}
 import org.elasticmq.msg.SendMessage
 import org.elasticmq.util.Logging
-import org.elasticmq.{MessageData, NewMessageData}
+import org.elasticmq.{MessageData, NewMessageData, QueueData, StringMessageAttribute}
 
 trait SendMessageOp extends Logging {
   this: QueueActorStorage =>
@@ -37,11 +37,30 @@ trait SendMessageOp extends Logging {
   }
 
   private def addMessage(message: NewMessageData) = {
-    val internalMessage = InternalMessage.from(message, queueData)
+    val updatedMessage = updateSequenceNumberAttribute(message, queueData)
+    val internalMessage = InternalMessage.from(updatedMessage, queueData)
     messageQueue += internalMessage
     fifoMessagesHistory = fifoMessagesHistory.addNew(internalMessage)
     logger.debug(s"${queueData.name}: Sent message with id ${internalMessage.id}")
 
     internalMessage.toMessageData
   }
+
+  private def updateSequenceNumberAttribute(
+      newMessageData: NewMessageData,
+      queueData: QueueData
+  ) = {
+    val updatedAttributes = {
+      if (!queueData.isFifo) {
+        (newMessageData.messageAttributes - "SequenceNumber")
+      } else {
+        newMessageData.messageAttributes.updated(
+          "SequenceNumber",
+          StringMessageAttribute(nextSequenceNumber().toString)
+        )
+      }
+    }
+    newMessageData.copy(messageAttributes = updatedAttributes)
+
+  }
 }
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ReceiveMessageDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ReceiveMessageDirectives.scala
index edfd3577..69a845ff 100644
--- a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ReceiveMessageDirectives.scala
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ReceiveMessageDirectives.scala
@@ -24,10 +24,11 @@ trait ReceiveMessageDirectives {
     val MessageDeduplicationIdAttribute = "MessageDeduplicationId"
     val MessageGroupIdAttribute = "MessageGroupId"
     val AWSTraceHeaderAttribute = "AWSTraceHeader"
+    val SequenceNumberAttribute = "SequenceNumber"
 
     val AllAttributeNames = SentTimestampAttribute :: ApproximateReceiveCountAttribute ::
       ApproximateFirstReceiveTimestampAttribute :: SenderIdAttribute :: MessageDeduplicationIdAttribute ::
-      MessageGroupIdAttribute :: AWSTraceHeaderAttribute :: Nil
+      MessageGroupIdAttribute :: AWSTraceHeaderAttribute :: SequenceNumberAttribute :: Nil
   }
 
   def receiveMessage(p: AnyParams) = {
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/SendMessageDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/SendMessageDirectives.scala
index 837b550a..355dd1cd 100644
--- a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/SendMessageDirectives.scala
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/SendMessageDirectives.scala
@@ -35,6 +35,12 @@ trait SendMessageDirectives { this: ElasticMQDirectives with SQSLimitsModule =>
                   {messageAttributeDigest.map(d => <MD5OfMessageAttributes>{d}</MD5OfMessageAttributes>).getOrElse(())}
                   <MD5OfMessageBody>{digest}</MD5OfMessageBody>
                   <MessageId>{message.id.id}</MessageId>
+                  {message.messageAttributes.get("SequenceNumber").flatMap(sn => {
+                    sn match {
+                      case StringMessageAttribute(x, _) => Some(<SequenceNumber>{x}</SequenceNumber>)
+                      case _ => None
+                    }
+                }).getOrElse(())}
                 </SendMessageResult>
                 <ResponseMetadata>
                   <RequestId>{EmptyRequestId}</RequestId>


Test patch:
diff --git a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
index e4ecac1c..72d2fcf8 100644
--- a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
+++ b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
@@ -834,6 +834,64 @@ class AmazonJavaSdkTestSuite extends SqsClientServerCommunication with Matchers
     m2 should be(Some("Body 3"))
   }
 
+  test("Message does not have SequenceNumber if it is not connected with FIFO queue") {
+    val queueUrl = client.createQueue(new CreateQueueRequest("testQueue1")).getQueueUrl
+
+    val sendingResult = client.sendMessage(new SendMessageRequest(queueUrl, "Body 1"))
+    val message = receiveSingleMessageObject(queueUrl).orNull
+
+    Option(sendingResult.getSequenceNumber).isEmpty should be(true)
+    message.getMessageAttributes.asScala.contains("SequenceNumber") should be(false)
+  }
+
+  test("Message should have SequenceNumber in result when sent to FIFO queue") {
+    val groupId = "1"
+    val queueUrl = createFifoQueue()
+
+    val sendingResult = client.sendMessage(new SendMessageRequest(queueUrl, "Body 1")
+      .withMessageGroupId(groupId))
+
+    Option(sendingResult.getSequenceNumber).isDefined should be(true)
+  }
+
+  test("FIFO queue - older message should have SequenceNumber which is less than that in newer message") {
+    val groupId = "1"
+    val queueUrl = createFifoQueue()
+
+    val firstSeqNum = client.sendMessage(new SendMessageRequest(queueUrl, "Body 1").withMessageGroupId(groupId)).getSequenceNumber
+    val secondSeqNum = client.sendMessage(new SendMessageRequest(queueUrl, "Body 2").withMessageGroupId(groupId)).getSequenceNumber
+
+    firstSeqNum.toLong should be <  secondSeqNum.toLong
+  }
+
+  test("FIFO queue - SequenceNumber continues to increase after deleting message from queue") {
+    val groupId1 = "1"
+    val queueUrl = createFifoQueue(attributes = Map(visibilityTimeoutAttribute -> "0"))
+
+    val seqNumFromDeleted = client.sendMessage(new SendMessageRequest(queueUrl, "Body 1").withMessageGroupId(groupId1)).getSequenceNumber
+
+    val willBeDeleted = client.receiveMessage(new ReceiveMessageRequest(queueUrl).withMessageAttributeNames("All")).getMessages.get(0)
+    client.deleteMessage(new DeleteMessageRequest(queueUrl, willBeDeleted.getReceiptHandle))
+
+    val secondSeqNum = client.sendMessage(new SendMessageRequest(queueUrl, "Body 2").withMessageGroupId(groupId1)).getSequenceNumber
+
+    seqNumFromDeleted.toLong should be < secondSeqNum.toLong
+  }
+
+  test("FIFO queue - SequenceNumber is not incremented between receives") {
+    val groupId1 = "1"
+    val queueUrl = createFifoQueue(attributes = Map(visibilityTimeoutAttribute -> "0"))
+    val res = client.sendMessage(new SendMessageRequest(queueUrl, "Body 1").withMessageGroupId(groupId1))
+
+    val seqNum1 = client.receiveMessage(new ReceiveMessageRequest(queueUrl).withMessageAttributeNames("All"))
+      .getMessages.get(0).getMessageAttributes.asScala("SequenceNumber").getStringValue
+    val seqNum2 = client.receiveMessage(new ReceiveMessageRequest(queueUrl).withMessageAttributeNames("All"))
+      .getMessages.get(0).getMessageAttributes.asScala("SequenceNumber").getStringValue
+
+    res.getSequenceNumber should equal(seqNum1)
+    res.getSequenceNumber should equal(seqNum2)
+  }
+
   def queueVisibilityTimeout(queueUrl: String): Long = getQueueLongAttribute(queueUrl, visibilityTimeoutAttribute)
 
   test("should receive no more than the given amount of messages") {