#259: Added function to allow processing messages at least once

zio-sqs/src/main/scala/zio/sqs/SqsStream.scala

@@ -2,7 +2,7 @@ package zio.sqs
 
 import zio.aws.sqs._
 import zio.aws.sqs.model._
-import zio.{ RIO, ZIO }
+import zio._
 import zio.stream.ZStream
 import zio.aws.sqs.model.primitives.MessageAttributeName
 
@@ -36,4 +36,52 @@ object SqsStream {
 
   def deleteMessage(queueUrl: String, msg: Message.ReadOnly): RIO[Sqs, Unit] =
     zio.aws.sqs.Sqs.deleteMessage(DeleteMessageRequest(queueUrl, msg.receiptHandle.getOrElse(""))).mapError(_.toThrowable)
+
+  /**
+   * Processes SQS messages from the queue using a given function and
+   * deletes the messages from the queue only if they processed successfully.
+   * This ensures that no message is lost in case of an error.
+   *
+   * If the message fails to be processed (=error or defect),
+   * it will reappear in the queue after the `visibilityTimeout` expires.
+   * If a DLQ is attached to the queue, the message will move into the DLQ after a set number of retries.
+   *
+   * Returns a stream original message and the outcome of the processing, to allow attaching logging or metrics.
+   * The returned Stream is also convenient for testing:
+   * exposing what was processed and allowing to stop the stream on some ocndition.
+   *
+   * @param url                 the URL of the SQS queue
+   * @param settings            the settings for the stream of messages from the queue
+   * @param parallelism         how many messages are processed in parallel
+   * @param deleteRetrySchedule the schedule by which to retry deleting the message (which should be rare)
+   * @param process             the function that takes an ObjectSummary and returns a ZIO effect
+   * @tparam E the error type of the ZIO effect
+   * @tparam A the value type of the ZIO effect
+   * @return a ZStream that can either fail with a Throwable or emit values of type Either[(Message.ReadOnly, Cause[E]), A]
+   */
+  def processMessages[E, A](
+    url: String,
+    settings: SqsStreamSettings,
+    parallelism: Int = 1,
+    deleteRetrySchedule: Schedule[Any, Any, Any] = defaultDeleteRetryPolicy
+  )(process: Message.ReadOnly => ZIO[Any, E, A]): ZStream[Sqs, Throwable, Either[(Message.ReadOnly, Cause[E]), A]] =
+    zio.sqs
+      .SqsStream(url, settings)
+      // Unordered since SQS (and queues in general) don't have ordering semantics
+      .mapZIOParUnordered(parallelism) { msg =>
+        process(msg)
+        // Capture any errors and defects
+        .sandbox.zipLeft {
+          SqsStream
+            .deleteMessage(url, msg)
+            // Retry in case it fails to delete transiently
+            .retry(deleteRetrySchedule)
+            // Die if it fails and stop the stream, to avoid reprocessing validly processed messages
+            // This would be very rare and caught early in testing.
+            .orDie
+        }.mapError(err => (msg, err)).either // Move the error to the return value, so that it doesn't stop the stream
+      }
+
+  private val defaultDeleteRetryPolicy: Schedule[Any, Any, Any] =
+    Schedule.exponential(10.milliseconds) && Schedule.upTo(30.seconds)
 }

zio-sqs/src/test/scala/zio/sqs/SqsConsumerSpec.scala

@@ -0,0 +1,40 @@
+package zio.sqs
+import zio.test._
+import zio._
+import zio.aws.sqs.Sqs
+import zio.aws.sqs.model.SendMessageRequest
+object SqsConsumerSpec extends ZIOSpecDefault {
+
+  def produce(url: String, message: String): ZIO[Sqs, Throwable, Unit] =
+    Sqs.sendMessage(SendMessageRequest(url, message)).unit.mapError(_.toThrowable)
+  override def spec: Spec[TestEnvironment with Scope, Any]             =
+    suite("SQS Consumer spec")(
+      test("retry failed messages indefinitely") {
+        val settings = SqsStreamSettings(
+          autoDelete = false,
+          stopWhenQueueEmpty = true,
+          visibilityTimeout = Some(1),
+          waitTimeSeconds = Some(2) // more than visibilityTimeout to allow errors to reappear
+        )
+        for {
+          _        <- Utils.createQueue("test-queue")
+          url      <- Utils.getQueueUrl("test-queue")
+          _        <- produce(url, "1")
+          _        <- produce(url, "not number1")
+          _        <- produce(url, "2")
+          messages <- SqsStream
+                        .processMessages(url, settings)(
+                          _.getBody.flatMap(body => ZIO.attempt(body.toInt))
+                        )
+                        .take(3 + 2)
+                        .runCollect
+          successes = messages.collect { case Right(v) => v }
+          failures  = messages.collect { case Left(v) => v }
+        } yield assertTrue(
+          successes == Chunk(1, 2),
+          failures.length == 3, // It is retried multiple times
+          failures.map(_._1.body.toOption.get).distinct.length == 1
+        )
+      }
+    ).provideLayer(MockSqsServerAndClient.layer)
+}

zio-sqs/src/test/scala/zio/sqs/ZioSqsMockServer.scala

@@ -1,14 +1,15 @@
 package zio.sqs
 
 import java.net.URI
-import zio.aws.core.config.AwsConfig
+import zio.aws.core.config.{ AwsConfig, CommonAwsConfig }
 import zio.aws.sqs.Sqs
 import org.elasticmq.rest.sqs.SQSRestServer
 import org.elasticmq.RelaxedSQSLimits
 import org.elasticmq.rest.sqs.TheSQSRestServerBuilder
 import org.elasticmq.NodeAddress
 import software.amazon.awssdk.auth.credentials.{ AwsBasicCredentials, StaticCredentialsProvider }
 import software.amazon.awssdk.regions.Region
+import zio.aws.netty.NettyHttpClient
 import zio.{ Scope, ZIO, ZLayer }
 
 object ZioSqsMockServer extends TheSQSRestServerBuilder(None, None, "", 9324, NodeAddress(), true, RelaxedSQSLimits, "elasticmq", "000000000000", None) {
@@ -31,3 +32,45 @@ object ZioSqsMockServer extends TheSQSRestServerBuilder(None, None, "", 9324, No
         .endpointOverride(uri)
     )
 }
+
+object MockSqsServerAndClient {
+
+  lazy val layer: ZLayer[Any, Throwable, Sqs] =
+    (NettyHttpClient.default ++ mockServer) >>> AwsConfig.configured() >>> zio.aws.sqs.Sqs.live
+
+  private lazy val mockServer: ZLayer[Any, Throwable, CommonAwsConfig] = {
+    val dummyAwsKeys =
+      StaticCredentialsProvider.create(AwsBasicCredentials.create("key", "key"))
+    val region       = Region.AP_NORTHEAST_2
+    ZLayer.scoped {
+      for {
+        // Random port to allow parallel tests
+        port          <- ZIO.randomWith(_.nextIntBetween(9000, 50000))
+        serverBuilder <- ZIO.attempt(
+                           TheSQSRestServerBuilder(
+                             providedActorSystem = None,
+                             providedQueueManagerActor = None,
+                             interface = "",
+                             port = port,
+                             serverAddress = NodeAddress(),
+                             generateServerAddress = true,
+                             sqsLimits = RelaxedSQSLimits,
+                             _awsRegion = region.toString,
+                             _awsAccountId = "000000000000",
+                             queueEventListener = None
+                           )
+                         )
+        server        <- ZIO.acquireRelease(
+                           ZIO.attempt(serverBuilder.start())
+                         )(server => ZIO.succeed(server.stopAndWait()))
+        awsConfig      = CommonAwsConfig(
+                           region = Some(region),
+                           credentialsProvider = dummyAwsKeys,
+                           endpointOverride = Some(new URI(s"http://localhost:${port}")),
+                           None
+                         )
+      } yield awsConfig
+    }
+  }
+
+}