Use AwsSessionCredentials to support aws oidc

aws/sqs/integration/README.md

@@ -0,0 +1,8 @@
+# How to run tests
+
+Tests are using [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-quickstart.html).
+Make sure to have the necessary permissions for accessing AWS resources: [AmazonSQSFullAccess](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AmazonSQSFullAccess.html).
+
+Steps:
+- `aws configure`
+- `sbt "project awsSqsIt" test`

aws/sqs/integration/src/test/scala/com/commercetools/queue/sqs/SqsClientSuite.scala

@@ -21,7 +21,7 @@ import cats.syntax.all._
 import com.commercetools.queue.QueueClient
 import com.commercetools.queue.aws.sqs.SQSClient
 import com.commercetools.queue.testkit.QueueClientSuite
-import software.amazon.awssdk.auth.credentials.{AnonymousCredentialsProvider, AwsBasicCredentials, StaticCredentialsProvider}
+import software.amazon.awssdk.auth.credentials.{AnonymousCredentialsProvider, AwsSessionCredentials, StaticCredentialsProvider}
 import software.amazon.awssdk.regions.Region
 
 import java.net.URI
@@ -40,9 +40,12 @@ class SqsClientSuite extends QueueClientSuite {
           .asScala
           .find(_.id == awsRegion)
           .liftTo[IO](new IllegalArgumentException(s"Cannot find any suitable AWS region from $awsRegion value!"))
-        credentials <- (string("AWS_SQS_ACCESS_KEY"), string("AWS_SQS_ACCESS_SECRET")).mapN((accessKey, accessSecret) =>
+        accessKey <- string("AWS_SQS_ACCESS_KEY")
+        accessSecret <- string("AWS_SQS_ACCESS_SECRET")
+        sessionToken <- string("AWS_SQS_SESSION_TOKEN")
+        credentials <- IO.pure(
           StaticCredentialsProvider.create(
-            AwsBasicCredentials.create(accessKey, accessSecret)
+            AwsSessionCredentials.create(accessKey, accessSecret, sessionToken)
           ))
       } yield (region, credentials, None)
     )

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

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)
 }

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,

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,

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

azure/service-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,21 @@ 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))
+      }
+    }
 }

azure/service-bus/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())
 
 }

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

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

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]

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

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)

docs/getting-started/subscribing.md

@@ -149,8 +149,8 @@ subscriber
         IO.println(s"Received $payload")
       }
     }.guaranteeCase {
-      case Outcome.Succeeded(_) => batch.ackAll
-      case _ => batch.nackAll
+      case Outcome.Succeeded(_) => batch.ackAll.as(())
+      case _ => batch.nackAll.as(())
     }
   }
 ```
@@ -253,8 +253,8 @@ subscriber.puller.use { queuePuller =>
         message.payload.flatMap { payload =>
           IO.println(s"Received $payload")
         }.guaranteeCase {
-          case Outcome.Succeeded(_) => batch.ackAll
-          case _ => batch.nackAll
+          case Outcome.Succeeded(_) => batch.ackAll.as(())
+          case _ => batch.nackAll.as(())
         }
       }
     }