Merge branch 'main' into metrics

commercetools · Apr 4, 2024 · f4358a1 · f4358a1
2 parents 5c6dd23 + 47d4969
commit f4358a1
Show file tree

Hide file tree

Showing 13 changed files with 231 additions and 32 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
@@ -41,21 +41,28 @@ class SQSAdministration[F[_]](client: SqsAsyncClient, getQueueUrl: String => F[S
             .build())
       }
     }.void
+      .adaptError(makeQueueException(_, name))
 
   override def delete(name: String): F[Unit] =
-    getQueueUrl(name).flatMap { queueUrl =>
-      F.fromCompletableFuture {
-        F.delay {
-          client.deleteQueue(
-            DeleteQueueRequest
-              .builder()
-              .queueUrl(queueUrl)
-              .build())
+    getQueueUrl(name)
+      .flatMap { queueUrl =>
+        F.fromCompletableFuture {
+          F.delay {
+            client.deleteQueue(
+              DeleteQueueRequest
+                .builder()
+                .queueUrl(queueUrl)
+                .build())
+          }
         }
       }
-    }.void
+      .void
+      .adaptError(makeQueueException(_, name))
 
   override def exists(name: String): F[Boolean] =
-    getQueueUrl(name).as(true).recover { case _: QueueDoesNotExistException => false }
+    getQueueUrl(name)
+      .as(true)
+      .recover { case _: QueueDoesNotExistException => false }
+      .adaptError(makeQueueException(_, name))
 
 }
diff --git a/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSClient.scala b/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSClient.scala
@@ -18,6 +18,7 @@ package com.commercetools.queue.aws.sqs
 
 import cats.effect.{Async, Resource}
 import cats.syntax.functor._
+import cats.syntax.monadError._
 import com.commercetools.queue.{Deserializer, QueueAdministration, QueueClient, QueuePublisher, QueueSubscriber, Serializer}
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider
 import software.amazon.awssdk.http.async.SdkAsyncHttpClient
@@ -35,6 +36,7 @@ class SQSClient[F[_]] private (client: SqsAsyncClient)(implicit F: Async[F]) ext
         client.getQueueUrl(GetQueueUrlRequest.builder().queueName(name).build())
       }
     }.map(_.queueUrl)
+      .adaptError(makeQueueException(_, name))
 
   override def administration: QueueAdministration[F] =
     new SQSAdministration(client, getQueueUrl(_))

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
@@ -18,7 +18,8 @@ package com.commercetools.queue.aws.sqs
 
 import cats.effect.Async
 import cats.syntax.functor._
-import com.commercetools.queue.MessageContext
+import cats.syntax.monadError._
+import com.commercetools.queue.{Action, MessageContext}
 import software.amazon.awssdk.services.sqs.SqsAsyncClient
 import software.amazon.awssdk.services.sqs.model.{ChangeMessageVisibilityRequest, DeleteMessageRequest}
 
@@ -31,6 +32,7 @@ class SQSMessageContext[F[_], T](
   receiptHandle: String,
   val messageId: String,
   lockTTL: Int,
+  queueName: String,
   queueUrl: String,
   client: SqsAsyncClient
 )(implicit F: Async[F])
@@ -42,6 +44,7 @@ class SQSMessageContext[F[_], T](
         client.deleteMessage(DeleteMessageRequest.builder().queueUrl(queueUrl).receiptHandle(receiptHandle).build())
       }
     }.void
+      .adaptError(makeMessageException(_, queueName, messageId, Action.Ack))
 
   override def nack(): F[Unit] =
     F.fromCompletableFuture {
@@ -55,6 +58,7 @@ class SQSMessageContext[F[_], T](
             .build())
       }
     }.void
+      .adaptError(makeMessageException(_, queueName, messageId, Action.Nack))
 
   override def extendLock(): F[Unit] =
     F.fromCompletableFuture {
@@ -68,5 +72,6 @@ class SQSMessageContext[F[_], T](
             .build())
       }
     }.void
+      .adaptError(makeMessageException(_, queueName, messageId, Action.ExtendLock))
 
 }
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
@@ -17,7 +17,10 @@
 package com.commercetools.queue.aws.sqs
 
 import cats.effect.Async
-import cats.syntax.all._
+import cats.syntax.either._
+import cats.syntax.flatMap._
+import cats.syntax.functor._
+import cats.syntax.monadError._
 import com.commercetools.queue.{Deserializer, MessageContext, QueuePuller}
 import fs2.Chunk
 import software.amazon.awssdk.services.sqs.SqsAsyncClient
@@ -51,19 +54,24 @@ class SQSPuller[F[_], T](
             .build())
       }
     }.flatMap { response =>
-      Chunk.iterator(response.messages().iterator().asScala).traverse { message =>
-        deserializer.deserialize(message.body()).liftTo[F].map { payload =>
-          new SQSMessageContext(
-            payload,
-            Instant.ofEpochMilli(message.attributes().get(MessageSystemAttributeName.SENT_TIMESTAMP).toLong),
-            message.attributesAsStrings().asScala.toMap,
-            message.receiptHandle(),
-            message.messageId(),
-            lockTTL,
-            queueUrl,
-            client
-          )
+      Chunk
+        .iterator(response.messages().iterator().asScala)
+        .traverse { message =>
+          deserializer.deserialize(message.body()).liftTo[F].map { payload =>
+            new SQSMessageContext(
+              payload = payload,
+              enqueuedAt =
+                Instant.ofEpochMilli(message.attributes().get(MessageSystemAttributeName.SENT_TIMESTAMP).toLong),
+              metadata = message.attributesAsStrings().asScala.toMap,
+              receiptHandle = message.receiptHandle(),
+              messageId = message.messageId(),
+              lockTTL = lockTTL,
+              queueName = queueName,
+              queueUrl = queueUrl,
+              client = client
+            )
+          }
         }
-      }
-    }
+    }.widen[Chunk[MessageContext[F, T]]]
+      .adaptError(makePullQueueException(_, queueName))
 }
diff --git a/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSPusher.scala b/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSPusher.scala
@@ -18,6 +18,7 @@ package com.commercetools.queue.aws.sqs
 
 import cats.effect.Async
 import cats.syntax.functor._
+import cats.syntax.monadError._
 import com.commercetools.queue.{QueuePusher, Serializer}
 import software.amazon.awssdk.services.sqs.SqsAsyncClient
 import software.amazon.awssdk.services.sqs.model.{SendMessageBatchRequest, SendMessageBatchRequestEntry, SendMessageRequest}
@@ -46,6 +47,7 @@ class SQSPusher[F[_], T](
             .build())
       }
     }.void
+      .adaptError(makePushQueueException(_, queueName))
 
   override def push(messages: List[T], delay: Option[FiniteDuration]): F[Unit] =
     F.fromCompletableFuture {
@@ -65,5 +67,6 @@ class SQSPusher[F[_], T](
             .build())
       }
     }.void
+      .adaptError(makePushQueueException(_, queueName))
 
 }
diff --git a/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSSubscriber.scala b/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSSubscriber.scala
@@ -42,6 +42,7 @@ class SQSSubscriber[F[_], T](
             .build())
       }
     }.map(_.attributes().get(QueueAttributeName.VISIBILITY_TIMEOUT).toInt)
+      .adaptError(makeQueueException(_, queueName))
 
   override def puller: Resource[F, QueuePuller[F, T]] =
     Resource.eval {

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
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2024 Commercetools GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.commercetools.queue.aws
+
+import com.commercetools.queue.{Action, CannotPullException, CannotPushException, MessageException, QueueAlreadyExistException, QueueDoesNotExistException, QueueException, UnknownQueueException}
+import software.amazon.awssdk.services.sqs.model.{QueueDoesNotExistException => AwsQueueDoesNotExistException, QueueNameExistsException}
+
+package object sqs {
+
+  def makeQueueException(t: Throwable, queueName: String): QueueException = t match {
+    case _: AwsQueueDoesNotExistException => QueueDoesNotExistException(queueName, t)
+    case _: QueueNameExistsException => QueueAlreadyExistException(queueName, t)
+    case t: QueueException => t
+    case _ => UnknownQueueException(queueName, t)
+  }
+
+  def makePushQueueException(t: Throwable, queueName: String): QueueException =
+    new CannotPushException(queueName, makeQueueException(t, queueName))
+
+  def makePullQueueException(t: Throwable, queueName: String): QueueException =
+    new CannotPullException(queueName, makeQueueException(t, queueName))
+
+  def makeMessageException(t: Throwable, queueName: String, msgId: String, action: Action): QueueException =
+    new MessageException(msgId = msgId, action = action, inner = makeQueueException(t, queueName))
+
+}
diff --git a/...us/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusAdministration.scala b/...us/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusAdministration.scala
@@ -18,6 +18,7 @@ package com.commercetools.queue.azure.servicebus
 
 import cats.effect.Async
 import cats.syntax.functor._
+import cats.syntax.monadError._
 import com.azure.messaging.servicebus.administration.ServiceBusAdministrationClient
 import com.azure.messaging.servicebus.administration.models.CreateQueueOptions
 import com.commercetools.queue.QueueAdministration
@@ -36,11 +37,16 @@ class ServiceBusAdministration[F[_]](client: ServiceBusAdministrationClient)(imp
           .setDefaultMessageTimeToLive(Duration.ofMillis(messageTTL.toMillis))
           .setLockDuration(Duration.ofMillis(lockTTL.toMillis))))
       .void
+      .adaptError(makeQueueException(_, name))
 
   override def delete(name: String): F[Unit] =
-    F.blocking(client.deleteQueue(name)).void
+    F.blocking(client.deleteQueue(name))
+      .void
+      .adaptError(makeQueueException(_, name))
 
   override def exists(name: String): F[Boolean] =
-    F.blocking(client.getQueueExists(name)).map(_.booleanValue)
+    F.blocking(client.getQueueExists(name))
+      .map(_.booleanValue)
+      .adaptError(makeQueueException(_, name))
 
 }
diff --git a/...ervice-bus/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusPuller.scala b/...ervice-bus/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusPuller.scala
@@ -16,8 +16,11 @@
 
 package com.commercetools.queue.azure.servicebus
 
-import cats.effect.kernel.Async
-import cats.syntax.all._
+import cats.effect.Async
+import cats.syntax.either._
+import cats.syntax.flatMap._
+import cats.syntax.functor._
+import cats.syntax.monadError._
 import com.azure.messaging.servicebus.ServiceBusReceiverClient
 import com.commercetools.queue.{Deserializer, MessageContext, QueuePuller}
 import fs2.Chunk
@@ -45,5 +48,7 @@ class ServiceBusPuller[F[_], Data](
           new ServiceBusMessageContext(data, sbMessage, receiver)
         })
     }
+    .widen[Chunk[MessageContext[F, Data]]]
+    .adaptError(makePullQueueException(_, queueName))
 
 }
diff --git a/...ervice-bus/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusPusher.scala b/...ervice-bus/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusPusher.scala
@@ -39,7 +39,9 @@ class ServiceBusPusher[F[_], Data](
       F.realTimeInstant
         .map(now => sbMessage.setScheduledEnqueueTime(now.plusMillis(delay.toMillis).atOffset(ZoneOffset.UTC)))
     } *>
-      F.blocking(sender.sendMessage(sbMessage)).void
+      F.blocking(sender.sendMessage(sbMessage))
+        .void
+        .adaptError(makePushQueueException(_, queueName))
   }
 
   override def push(messages: List[Data], delay: Option[FiniteDuration]): F[Unit] = {
@@ -51,7 +53,9 @@ class ServiceBusPusher[F[_], Data](
         }
       }
     } *>
-      F.blocking(sender.sendMessages(sbMessages.asJava)).void
+      F.blocking(sender.sendMessages(sbMessages.asJava))
+        .void
+        .adaptError(makePushQueueException(_, queueName))
   }
 
 }
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
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2024 Commercetools GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.commercetools.queue.azure
+
+import com.azure.core.exception.{ResourceExistsException, ResourceNotFoundException}
+import com.commercetools.queue.{Action, CannotPullException, CannotPushException, MessageException, QueueAlreadyExistException, QueueDoesNotExistException, QueueException, UnknownQueueException}
+
+package object servicebus {
+
+  def makeQueueException(t: Throwable, queueName: String): QueueException =
+    t match {
+      case _: ResourceNotFoundException => QueueDoesNotExistException(queueName, t)
+      case _: ResourceExistsException => QueueAlreadyExistException(queueName, t)
+      case t: QueueException => t
+      case _ => UnknownQueueException(queueName, t)
+    }
+
+  def makePushQueueException(t: Throwable, queueName: String): QueueException =
+    new CannotPushException(queueName, makeQueueException(t, queueName))
+
+  def makePullQueueException(t: Throwable, queueName: String): QueueException =
+    new CannotPullException(queueName, makeQueueException(t, queueName))
+
+  def makeMessageException(t: Throwable, queueName: String, msgId: String, action: Action): QueueException =
+    new MessageException(msgId = msgId, action = action, inner = makeQueueException(t, queueName))
+
+}
diff --git a/core/src/main/scala/com/commercetools/queue/Action.scala b/core/src/main/scala/com/commercetools/queue/Action.scala
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2024 Commercetools GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.commercetools.queue
+
+import cats.Show
+
+sealed trait Action
+
+object Action {
+  case object Ack extends Action
+  case object Nack extends Action
+  case object ExtendLock extends Action
+
+  implicit val show: Show[Action] = Show.show {
+    case Ack => "ack"
+    case Nack => "nack"
+    case ExtendLock => "extend lock"
+  }
+}