Return List[MessageId] on ackAll nackAll

commercetools · Oct 29, 2024 · 38bd12b · 38bd12b
1 parent ad85e0b
commit 38bd12b
Show file tree

Hide file tree

Showing 22 changed files with 157 additions and 113 deletions.
diff --git a/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSAdministration.scala b/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSAdministration.scala
@@ -43,7 +43,9 @@ private class SQSAdministration[F[_]](client: SqsAsyncClient, getQueueUrl: Strin
             .queueName(name)
             .attributes(Map(
               QueueAttributeName.MESSAGE_RETENTION_PERIOD -> messageTTL.toSeconds.toString(),
-              QueueAttributeName.VISIBILITY_TIMEOUT -> lockTTL.toSeconds.toString()).asJava)
+              QueueAttributeName.VISIBILITY_TIMEOUT -> lockTTL.toSeconds.toString(),
+              QueueAttributeName.RECEIVE_MESSAGE_WAIT_TIME_SECONDS -> "20" // this is meant to enable long polling https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-configure-queue-parameters.html (see: Receive message wait time)
+            ).asJava)
             .build())
       }
     }.void

diff --git a/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSMessageBatch.scala b/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSMessageBatch.scala
@@ -18,10 +18,12 @@ package com.commercetools.queue.aws.sqs
 
 import cats.effect.Async
 import cats.implicits.toFunctorOps
-import com.commercetools.queue.{Message, UnsealedMessageBatch}
+import com.commercetools.queue.{Message, MessageId, UnsealedMessageBatch}
 import fs2.Chunk
 import software.amazon.awssdk.services.sqs.SqsAsyncClient
-import software.amazon.awssdk.services.sqs.model.{ChangeMessageVisibilityBatchRequest, ChangeMessageVisibilityBatchRequestEntry, DeleteMessageBatchRequest, DeleteMessageBatchRequestEntry}
+import software.amazon.awssdk.services.sqs.model._
+
+import scala.jdk.CollectionConverters.CollectionHasAsScala
 
 private class SQSMessageBatch[F[_], T](
   payload: Chunk[SQSMessageContext[F, T]],
@@ -32,7 +34,7 @@ private class SQSMessageBatch[F[_], T](
 
   override def messages: Chunk[Message[F, T]] = payload
 
-  override def ackAll: F[Unit] =
+  override def ackAll: F[List[MessageId]] =
     F.fromCompletableFuture {
       F.delay {
         client.deleteMessageBatch(
@@ -43,33 +45,33 @@ private class SQSMessageBatch[F[_], T](
               DeleteMessageBatchRequestEntry
                 .builder()
                 .receiptHandle(m.receiptHandle)
-                .id(m.messageId)
+                .id(m.messageId.value)
                 .build()
             }.asJava)
             .build()
         )
       }
-    }.void
+    }.map(res => res.failed().asScala.map(message => MessageId(message.id())).toList)
 
-  override def nackAll: F[Unit] = F.fromCompletableFuture {
-    F.delay {
-      val req = ChangeMessageVisibilityBatchRequest
-        .builder()
-        .queueUrl(queueUrl)
-        .entries(
-          payload.map { m =>
-            ChangeMessageVisibilityBatchRequestEntry
-              .builder()
-              .id(m.messageId)
-              .receiptHandle(m.receiptHandle)
-              .visibilityTimeout(0)
-              .build()
-          }.asJava
+  override def nackAll: F[List[MessageId]] =
+    F.fromCompletableFuture {
+      F.delay {
+        client.changeMessageVisibilityBatch(
+          ChangeMessageVisibilityBatchRequest
+            .builder()
+            .queueUrl(queueUrl)
+            .entries(
+              payload.map { m =>
+                ChangeMessageVisibilityBatchRequestEntry
+                  .builder()
+                  .id(m.messageId.value)
+                  .receiptHandle(m.receiptHandle)
+                  .visibilityTimeout(0)
+                  .build()
+              }.asJava
+            )
+            .build()
         )
-        .build()
-      client.changeMessageVisibilityBatch(
-        req
-      )
-    }
-  }.void
+      }
+    }.map(res => res.failed().asScala.map(message => MessageId(message.id())).toList)
 }
diff --git a/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSMessageContext.scala b/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSMessageContext.scala
@@ -19,7 +19,7 @@ package com.commercetools.queue.aws.sqs
 import cats.effect.Async
 import cats.syntax.functor._
 import cats.syntax.monadError._
-import com.commercetools.queue.{Action, UnsealedMessageContext}
+import com.commercetools.queue.{Action, MessageId, UnsealedMessageContext}
 import software.amazon.awssdk.services.sqs.SqsAsyncClient
 import software.amazon.awssdk.services.sqs.model.{ChangeMessageVisibilityRequest, DeleteMessageRequest}
 
@@ -31,7 +31,7 @@ private class SQSMessageContext[F[_], T](
   val enqueuedAt: Instant,
   val metadata: Map[String, String],
   val receiptHandle: String,
-  val messageId: String,
+  val messageId: MessageId,
   lockTTL: Int,
   queueName: String,
   queueUrl: String,

diff --git a/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSPuller.scala b/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSPuller.scala
@@ -21,7 +21,7 @@ import cats.effect.syntax.concurrent._
 import cats.syntax.flatMap._
 import cats.syntax.functor._
 import cats.syntax.monadError._
-import com.commercetools.queue.{Deserializer, MessageBatch, MessageContext, UnsealedQueuePuller}
+import com.commercetools.queue.{Deserializer, MessageBatch, MessageContext, MessageId, UnsealedQueuePuller}
 import fs2.Chunk
 import software.amazon.awssdk.services.sqs.SqsAsyncClient
 import software.amazon.awssdk.services.sqs.model.{MessageSystemAttributeName, ReceiveMessageRequest}
@@ -83,7 +83,7 @@ private class SQSPuller[F[_], T](
                   .collect { case (k, v) if v.dataType() == "String" => (k, v.stringValue()) }
                   .toMap,
                 receiptHandle = message.receiptHandle(),
-                messageId = message.messageId(),
+                messageId = MessageId(message.messageId()),
                 lockTTL = lockTTL,
                 queueName = queueName,
                 queueUrl = queueUrl,

diff --git a/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/package.scala b/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/package.scala
@@ -16,7 +16,7 @@
 
 package com.commercetools.queue.aws
 
-import com.commercetools.queue.{Action, CannotPullException, CannotPushException, MessageException, QueueAlreadyExistException, QueueDoesNotExistException, QueueException, UnknownQueueException}
+import com.commercetools.queue.{Action, CannotPullException, CannotPushException, MessageException, MessageId, QueueAlreadyExistException, QueueDoesNotExistException, QueueException, UnknownQueueException}
 import software.amazon.awssdk.services.sqs.model.{QueueDoesNotExistException => AwsQueueDoesNotExistException, QueueNameExistsException}
 
 package object sqs {
@@ -37,7 +37,7 @@ package object sqs {
       case _ => new CannotPullException(queueName, makeQueueException(t, queueName))
     }
 
-  private[sqs] def makeMessageException(t: Throwable, queueName: String, msgId: String, action: Action)
+  private[sqs] def makeMessageException(t: Throwable, queueName: String, msgId: MessageId, action: Action)
     : QueueException =
     t match {
       case t: QueueException => t

diff --git a/...-bus/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusMessageBatch.scala b/...-bus/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusMessageBatch.scala
@@ -17,8 +17,9 @@
 package com.commercetools.queue.azure.servicebus
 
 import cats.effect.Async
+import cats.implicits.{catsSyntaxApplicativeError, toFlatMapOps, toFunctorOps}
 import com.azure.messaging.servicebus.ServiceBusReceiverClient
-import com.commercetools.queue.{Message, UnsealedMessageBatch}
+import com.commercetools.queue.{Message, MessageId, UnsealedMessageBatch}
 import fs2.Chunk
 
 private class ServiceBusMessageBatch[F[_], T](
@@ -28,11 +29,17 @@ private class ServiceBusMessageBatch[F[_], T](
   extends UnsealedMessageBatch[F, T] {
   override def messages: Chunk[Message[F, T]] = payload
 
-  override def ackAll: F[Unit] = F.blocking {
-    payload.foreach(mCtx => receiver.complete(mCtx.underlying))
-  }
+  override def ackAll: F[List[MessageId]] =
+    payload.toList.foldLeft(F.pure(List[MessageId]())) { (accF, mCtx) =>
+      accF.flatMap { acc =>
+        F.pure(receiver.complete(mCtx.underlying)).as(acc).handleError(_ => acc :+ MessageId(mCtx.underlying.getMessageId))
+      }
+    }
 
-  override def nackAll: F[Unit] = F.blocking {
-    payload.foreach(mCtx => receiver.abandon(mCtx.underlying))
-  }
+  override def nackAll: F[List[MessageId]] =
+    payload.toList.foldLeft(F.pure(List[MessageId]())) { (accF, mCtx) =>
+      accF.flatMap { acc =>
+        F.pure(receiver.abandon(mCtx.underlying)).as(acc).handleError(_ => acc :+ MessageId(mCtx.underlying.getMessageId))
+      }
+    }
 }
diff --git a/...us/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusMessageContext.scala b/...us/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusMessageContext.scala
@@ -19,7 +19,7 @@ package com.commercetools.queue.azure.servicebus
 import cats.effect.Async
 import cats.syntax.functor._
 import com.azure.messaging.servicebus.{ServiceBusReceivedMessage, ServiceBusReceiverClient}
-import com.commercetools.queue.UnsealedMessageContext
+import com.commercetools.queue.{MessageId, UnsealedMessageContext}
 
 import java.time.Instant
 import scala.jdk.CollectionConverters.MapHasAsScala
@@ -49,6 +49,6 @@ private class ServiceBusMessageContext[F[_], T](
   override def extendLock(): F[Unit] =
     F.blocking(receiver.renewMessageLock(underlying)).void
 
-  override val messageId: String = underlying.getMessageId()
+  override val messageId: MessageId = MessageId(underlying.getMessageId())
 
 }
diff --git a/azure/service-bus/src/main/scala/com/commercetools/queue/azure/servicebus/package.scala b/azure/service-bus/src/main/scala/com/commercetools/queue/azure/servicebus/package.scala
@@ -17,7 +17,7 @@
 package com.commercetools.queue.azure
 
 import com.azure.core.exception.{ResourceExistsException, ResourceNotFoundException}
-import com.commercetools.queue.{Action, CannotPullException, CannotPushException, MessageException, QueueAlreadyExistException, QueueDoesNotExistException, QueueException, UnknownQueueException}
+import com.commercetools.queue.{Action, CannotPullException, CannotPushException, MessageException, MessageId, QueueAlreadyExistException, QueueDoesNotExistException, QueueException, UnknownQueueException}
 
 package object servicebus {
 
@@ -38,7 +38,7 @@ package object servicebus {
       case _ => new CannotPullException(queueName, makeQueueException(t, queueName))
     }
 
-  private[servicebus] def makeMessageException(t: Throwable, queueName: String, msgId: String, action: Action)
+  private[servicebus] def makeMessageException(t: Throwable, queueName: String, msgId: MessageId, action: Action)
     : QueueException =
     t match {
       case t: QueueException => t

diff --git a/core/src/main/scala/com/commercetools/queue/Message.scala b/core/src/main/scala/com/commercetools/queue/Message.scala
@@ -18,6 +18,8 @@ package com.commercetools.queue
 
 import java.time.Instant
 
+case class MessageId(value: String) extends AnyVal
+
 /**
  * Interface to access message data received from a queue.
  */
@@ -26,7 +28,7 @@ trait Message[F[_], T] {
   /**
    * Unique message identifier
    */
-  def messageId: String
+  def messageId: MessageId
 
   /**
    * The deserialized message payload

diff --git a/core/src/main/scala/com/commercetools/queue/MessageBatch.scala b/core/src/main/scala/com/commercetools/queue/MessageBatch.scala
@@ -33,12 +33,12 @@ sealed trait MessageBatch[F[_], T] {
   /**
    * Acknowledges all the messages in the chunk.
    */
-  def ackAll: F[Unit]
+  def ackAll: F[List[MessageId]]
 
   /**
    * Mark all messages from the chunk as non acknowledged.
    */
-  def nackAll: F[Unit]
+  def nackAll: F[List[MessageId]]
 }
 
 private[queue] trait UnsealedMessageBatch[F[_], T] extends MessageBatch[F, T]
diff --git a/core/src/main/scala/com/commercetools/queue/QueuePuller.scala b/core/src/main/scala/com/commercetools/queue/QueuePuller.scala
@@ -32,7 +32,7 @@ sealed trait QueuePuller[F[_], T] {
    * Pulls one batch of messages from the underlying queue system.
    *
    * The method returns chunks of size `batchSize` max, and (semantically)
-   * blocks for `waitingTime` before returning. The chunk might be empty
+   * blocks for `waitingTime` before returning any messages. The chunk might be empty
    * if no new messages are available during the waiting time.
    *
    * '''Note:''' the messages returned by this method must be manually

diff --git a/core/src/main/scala/com/commercetools/queue/errors.scala b/core/src/main/scala/com/commercetools/queue/errors.scala
@@ -40,8 +40,8 @@ case class CannotPushException(name: String, inner: Throwable)
 case class CannotPullException(name: String, inner: Throwable)
   extends QueueException(show"Cannot pull messages from queue $name", inner)
 
-case class MessageException(msgId: String, action: Action, inner: Throwable)
-  extends QueueException(show"Cannot $action message $msgId", inner)
+case class MessageException(msgId: MessageId, action: Action, inner: Throwable)
+  extends QueueException(show"Cannot $action message ${msgId.value}", inner)
 
 case class UnknownQueueException(name: String, inner: Throwable)
   extends QueueException(show"Something went wrong when interacting with queue $name", inner)

diff --git a/gcp/pubsub/src/main/scala/com/commercetools/queue/gcp/pubsub/PubSubMessageBatch.scala b/gcp/pubsub/src/main/scala/com/commercetools/queue/gcp/pubsub/PubSubMessageBatch.scala
@@ -17,12 +17,14 @@
 package com.commercetools.queue.gcp.pubsub
 
 import cats.effect.Async
-import cats.implicits.toFunctorOps
-import com.commercetools.queue.{Message, UnsealedMessageBatch}
+import cats.implicits.{catsSyntaxApplicativeError, toFlatMapOps, toFunctorOps}
+import com.commercetools.queue.{Message, MessageId, UnsealedMessageBatch}
 import com.google.cloud.pubsub.v1.stub.SubscriberStub
 import com.google.pubsub.v1.{AcknowledgeRequest, ModifyAckDeadlineRequest, SubscriptionName}
 import fs2.Chunk
 
+import scala.jdk.CollectionConverters.IterableHasAsJava
+
 private class PubSubMessageBatch[F[_], T](
   payload: Chunk[PubSubMessageContext[F, T]],
   subscriptionName: SubscriptionName,
@@ -31,27 +33,41 @@ private class PubSubMessageBatch[F[_], T](
   extends UnsealedMessageBatch[F, T] {
   override def messages: Chunk[Message[F, T]] = payload
 
-  override def ackAll: F[Unit] =
-    wrapFuture(
-      F.delay(
-        subscriber
-          .acknowledgeCallable()
-          .futureCall(
-            AcknowledgeRequest
-              .newBuilder()
-              .setSubscription(subscriptionName.toString)
-              .addAllAckIds(payload.map(_.underlying.getAckId).asJava)
-              .build()))).void
+  override def ackAll: F[List[MessageId]] =
+    payload.toList.foldLeft(F.pure(List[MessageId]())) { (accF, message) =>
+      accF.flatMap { acc =>
+        F.delay {
+          subscriber
+            .acknowledgeCallable()
+            .futureCall(
+              AcknowledgeRequest
+                .newBuilder()
+                .setSubscription(subscriptionName.toString)
+                .addAllAckIds(List(message.underlying.getAckId).asJava)
+                .build()
+            )
+        }.as(acc)
+          .handleError(_ => acc :+ MessageId(message.underlying.getMessage.getMessageId))
+      }
+    }
+
+  override def nackAll: F[List[MessageId]] =
+    payload.toList.foldLeft(F.pure(List[MessageId]())) { (accF, message) =>
+      accF.flatMap { acc =>
+        F.delay {
+          subscriber
+            .modifyAckDeadlineCallable()
+            .futureCall(
+              ModifyAckDeadlineRequest
+                .newBuilder()
+                .setSubscription(subscriptionName.toString)
+                .setAckDeadlineSeconds(0)
+                .addAllAckIds(List(message.underlying.getAckId).asJava)
+                .build()
+            )
+        }.as(acc)
+          .handleError(_ => acc :+ MessageId(message.underlying.getMessage.getMessageId))
+      }
+    }
 
-  override def nackAll: F[Unit] = wrapFuture(
-    F.delay(
-      subscriber
-        .modifyAckDeadlineCallable()
-        .futureCall(
-          ModifyAckDeadlineRequest
-            .newBuilder()
-            .setSubscription(subscriptionName.toString)
-            .setAckDeadlineSeconds(0)
-            .addAllAckIds(payload.map(_.underlying.getAckId).asJava)
-            .build()))).void
 }
diff --git a/gcp/pubsub/src/main/scala/com/commercetools/queue/gcp/pubsub/PubSubMessageContext.scala b/gcp/pubsub/src/main/scala/com/commercetools/queue/gcp/pubsub/PubSubMessageContext.scala
@@ -19,7 +19,7 @@ package com.commercetools.queue.gcp.pubsub
 import cats.effect.Async
 import cats.syntax.functor._
 import cats.syntax.monadError._
-import com.commercetools.queue.{Action, UnsealedMessageContext}
+import com.commercetools.queue.{Action, MessageId, UnsealedMessageContext}
 import com.google.cloud.pubsub.v1.stub.SubscriberStub
 import com.google.pubsub.v1.{AcknowledgeRequest, ModifyAckDeadlineRequest, ReceivedMessage, SubscriptionName}
 
@@ -36,7 +36,7 @@ private class PubSubMessageContext[F[_], T](
 )(implicit F: Async[F])
   extends UnsealedMessageContext[F, T] {
 
-  override def messageId: String = underlying.getAckId()
+  override def messageId: MessageId = MessageId(underlying.getAckId())
 
   override def rawPayload: String = underlying.getMessage().getData().toStringUtf8()
 

diff --git a/gcp/pubsub/src/main/scala/com/commercetools/queue/gcp/pubsub/package.scala b/gcp/pubsub/src/main/scala/com/commercetools/queue/gcp/pubsub/package.scala
@@ -19,7 +19,7 @@ package com.commercetools.queue.gcp
 import cats.effect.Async
 import cats.syntax.either._
 import cats.syntax.functor._
-import com.commercetools.queue.{Action, CannotPullException, CannotPushException, MessageException, QueueAlreadyExistException, QueueDoesNotExistException, QueueException, UnknownQueueException}
+import com.commercetools.queue.{Action, CannotPullException, CannotPushException, MessageException, MessageId, QueueAlreadyExistException, QueueDoesNotExistException, QueueException, UnknownQueueException}
 import com.google.api.core.{ApiFuture, ApiFutureCallback, ApiFutures}
 import com.google.api.gax.rpc.{AlreadyExistsException, NotFoundException}
 import com.google.common.util.concurrent.MoreExecutors
@@ -69,7 +69,7 @@ package object pubsub {
       case _ => new CannotPullException(queueName, makeQueueException(t, queueName))
     }
 
-  private[pubsub] def makeMessageException(t: Throwable, queueName: String, msgId: String, action: Action)
+  private[pubsub] def makeMessageException(t: Throwable, queueName: String, msgId: MessageId, action: Action)
     : QueueException =
     t match {
       case t: QueueException => t