AWS SQS: Keep message order (#1214)

This fixes messages arriving order when pulling from a FIFO queue and adds test coverage for that particular scenario. Cheers!
akka · Sep 17, 2018 · 26fe3d3 · 26fe3d3
1 parent fe62bd1
commit 26fe3d3
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 2 deletions.
diff --git a/sqs/src/main/scala/akka/stream/alpakka/sqs/SqsSourceStage.scala b/sqs/src/main/scala/akka/stream/alpakka/sqs/SqsSourceStage.scala
@@ -75,7 +75,7 @@ final class SqsSourceStage(queueUrl: String, settings: SqsSourceSettings)(implic
         currentRequests = currentRequests - 1
         maxCurrentConcurrency = if (result.getMessages.isEmpty) 1 else maxConcurrency
 
-        val receivedMessages = result.getMessages.asScala.reverse
+        val receivedMessages = result.getMessages.asScala
         receivedMessages.foreach(buffer.offer)
 
         if (receivedMessages.isEmpty && settings.closeOnEmptyReceive) {

diff --git a/sqs/src/test/scala/akka/stream/alpakka/sqs/scaladsl/DefaultTestContext.scala b/sqs/src/test/scala/akka/stream/alpakka/sqs/scaladsl/DefaultTestContext.scala
@@ -7,6 +7,7 @@ package akka.stream.alpakka.sqs.scaladsl
 import akka.actor.ActorSystem
 import akka.stream.{ActorMaterializer, Materializer}
 import com.amazonaws.services.sqs.AmazonSQSAsync
+import com.amazonaws.services.sqs.model.CreateQueueRequest
 import org.elasticmq.rest.sqs.{SQSLimits, SQSRestServer, SQSRestServerBuilder}
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Suite, Tag}
@@ -33,6 +34,12 @@ trait DefaultTestContext extends BeforeAndAfterAll with BeforeAndAfterEach with
 
   def randomQueueUrl(): String = sqsClient.createQueue(s"queue-${Random.nextInt}").getQueueUrl
 
+  val fifoQueueRequest = new CreateQueueRequest(s"queue-${Random.nextInt}.fifo")
+    .addAttributesEntry("FifoQueue", "true")
+    .addAttributesEntry("ContentBasedDeduplication", "true")
+
+  def randomFifoQueueUrl(): String = sqsClient.createQueue(fifoQueueRequest).getQueueUrl
+
   override protected def beforeEach(): Unit =
     sqsServer = SQSRestServerBuilder.withActorSystem(system).withSQSLimits(SQSLimits.Relaxed).withDynamicPort().start()
 
@@ -62,5 +69,4 @@ trait DefaultTestContext extends BeforeAndAfterAll with BeforeAndAfterEach with
     //#init-client
     awsSqsClient
   }
-
 }
diff --git a/sqs/src/test/scala/akka/stream/alpakka/sqs/scaladsl/SqsSpec.scala b/sqs/src/test/scala/akka/stream/alpakka/sqs/scaladsl/SqsSpec.scala
@@ -188,6 +188,26 @@ class SqsSpec extends FlatSpec with Matchers with DefaultTestContext {
     probe.cancel()
   }
 
+  it should "pull messages from a fifo queue following the same production order" taggedAs Integration in {
+    val queue = randomFifoQueueUrl()
+    implicit val awsSqsClient = sqsClient
+
+    for (i <- 0 until 10) {
+      val msg = s"Message - $i"
+      awsSqsClient.sendMessage(new SendMessageRequest(queue, msg).withMessageGroupId("group1"))
+    }
+
+    val probe = SqsSource(queue, sqsSourceSettings)
+      .runWith(TestSink.probe[Message])
+
+    for (i <- 0 until 10) {
+      val result = probe.requestNext()
+      result.getBody shouldBe s"Message - $i"
+    }
+
+    probe.cancel()
+  }
+
   it should "publish batch of messages and pull them" taggedAs Integration in {
     val queue = randomQueueUrl()
     implicit val awsSqsClient = sqsClient