Merge pull request #24 from commercetools/message-handler

Introducing a message handler abstraction

build.sbt

@@ -43,7 +43,11 @@ lazy val core = crossProject(JVMPlatform)
   .in(file("core"))
   .settings(commonSettings)
   .settings(
-    name := "fs2-queues-core"
+    name := "fs2-queues-core",
+    // TODO: Remove once 0.3 is published
+    mimaBinaryIssueFilters ++= List(
+      ProblemFilters.exclude[DirectMissingMethodProblem]("com.commercetools.queue.QueueSubscriber.this")
+    )
   )
 
 lazy val testkit = crossProject(JVMPlatform)
@@ -85,6 +89,10 @@ lazy val otel4s = crossProject(JVMPlatform)
     description := "Support for metrics and tracing using otel4s",
     libraryDependencies ++= List(
       "org.typelevel" %%% "otel4s-core" % "0.7.0"
+    ),
+    // TODO: Remove once 0.3 is published
+    mimaBinaryIssueFilters ++= List(
+      ProblemFilters.exclude[DirectMissingMethodProblem]("com.commercetools.queue.otel4s.MeasuringQueueSubscriber.this")
     )
   )
   .dependsOn(core % "compile->compile;test->test")
@@ -109,6 +117,11 @@ lazy val azureServiceBus = crossProject(JVMPlatform)
     name := "fs2-queues-azure-service-bus",
     libraryDependencies ++= List(
       "com.azure" % "azure-messaging-servicebus" % "7.17.0"
+    ),
+    // TODO: Remove once 0.3 is published
+    mimaBinaryIssueFilters ++= List(
+      ProblemFilters.exclude[DirectMissingMethodProblem](
+        "com.commercetools.queue.azure.servicebus.ServiceBusQueueSubscriber.this")
     )
   )
   .dependsOn(core, testkit % Test)
@@ -121,6 +134,10 @@ lazy val awsSQS = crossProject(JVMPlatform)
     name := "fs2-queues-aws-sqs",
     libraryDependencies ++= List(
       "software.amazon.awssdk" % "sqs" % "2.25.50"
+    ),
+    // TODO: Remove once 0.3 is published
+    mimaBinaryIssueFilters ++= List(
+      ProblemFilters.exclude[DirectMissingMethodProblem]("com.commercetools.queue.aws.sqs.SQSSubscriber.this")
     )
   )
   .dependsOn(core)

core/src/main/scala/com/commercetools/queue/MessageHandler.scala

@@ -0,0 +1,48 @@
+/*
+ * 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.MonadThrow
+import cats.syntax.applicativeError._
+import cats.syntax.functor._
+
+import scala.concurrent.duration.FiniteDuration
+
+trait MessageHandler[F[_], T, Res, D[_] <: Decision[_]] {
+  def handle(msg: Message[F, T]): F[D[Res]]
+}
+
+sealed trait Decision[+O]
+sealed trait ImmediateDecision[+O] extends Decision[O]
+object Decision {
+  case class Ok[O](res: O) extends ImmediateDecision[O]
+  case object Drop extends ImmediateDecision[Nothing]
+  case class Fail(t: Throwable, ack: Boolean) extends ImmediateDecision[Nothing]
+  case class Reenqueue(metadata: Option[Map[String, String]] = None, delay: Option[FiniteDuration] = None)
+    extends Decision[Nothing]
+}
+
+object MessageHandler {
+  // nack on any failure except for deserialization exception
+  def default[F[_]: MonadThrow, T, O](f: Message[F, T] => F[O]): MessageHandler[F, T, O, ImmediateDecision] =
+    msg =>
+      f(msg).attempt.map {
+        case Left(de: DeserializationException) => Decision.Fail(de, ack = true)
+        case Left(t) => Decision.Fail(t, ack = false)
+        case Right(a) => Decision.Ok(a)
+      }
+}

core/src/main/scala/com/commercetools/queue/QueuePublisher.scala

@@ -49,3 +49,14 @@ abstract class QueuePublisher[F[_], T](implicit F: MonadCancel[F, Throwable]) {
     }
 
 }
+
+object QueuePublisher {
+
+  /**
+   * A publisher that does nothing.
+   */
+  def noop[F[_], T](implicit F: MonadCancel[F, Throwable]) = new QueuePublisher[F, T] {
+    override def queueName: String = ""
+    override def pusher: Resource[F, QueuePusher[F, T]] = Resource.pure(QueuePusher.noop)
+  }
+}

core/src/main/scala/com/commercetools/queue/QueuePusher.scala

@@ -16,6 +16,8 @@
 
 package com.commercetools.queue
 
+import cats.Applicative
+
 import scala.concurrent.duration.FiniteDuration
 
 /**
@@ -38,3 +40,15 @@ trait QueuePusher[F[_], T] {
   def push(messages: List[(T, Map[String, String])], delay: Option[FiniteDuration]): F[Unit]
 
 }
+
+object QueuePusher {
+
+  /**
+   * A pusher that does nothing.
+   */
+  def noop[F[_], T](implicit F: Applicative[F]) = new QueuePusher[F, T] {
+    override def queueName: String = ""
+    override def push(message: T, metadata: Map[String, String], delay: Option[FiniteDuration]): F[Unit] = F.unit
+    override def push(messages: List[(T, Map[String, String])], delay: Option[FiniteDuration]): F[Unit] = F.unit
+  }
+}

core/src/main/scala/com/commercetools/queue/QueueSubscriber.scala

@@ -126,4 +126,52 @@ abstract class QueueSubscriber[F[_], T](implicit F: Concurrent[F]) {
         case Left(_) => ctx.nack()
       })
 
+  /**
+   * Processes the messages with the provided message handler.
+   * The messages are ack'ed, nack'ed or reenqueu'ed based on the decision returned from the handler.
+   * The stream emits results or errors down-stream and does not fail on business logic errors,
+   * allowing you to build error recovery logic.
+   *
+   * Messages in a batch are processed in parallel but result is emitted in order the messages were received,
+   * with the exclusion of the messages that have been reenqueu'ed and dropped.
+   */
+  final def process[Res](
+    batchSize: Int,
+    waitingTime: FiniteDuration,
+    publisherForReenqueue: QueuePublisher[F, T]
+  )(handler: MessageHandler[F, T, Res, Decision]
+  ): Stream[F, Either[Throwable, Res]] =
+    Stream
+      .resource(publisherForReenqueue.pusher)
+      .flatMap { pusher =>
+        messages(batchSize, waitingTime)
+          .parEvalMap(batchSize) { ctx =>
+            handler.handle(ctx).flatMap[Option[Either[Throwable, Res]]] {
+              case Decision.Ok(res) => ctx.ack().as(res.asRight.some)
+              case Decision.Drop => ctx.ack().as(none)
+              case Decision.Fail(t, true) => ctx.ack().as(t.asLeft.some)
+              case Decision.Fail(t, false) => ctx.nack().as(t.asLeft.some)
+              case Decision.Reenqueue(metadata, delay) =>
+                ctx.payload.flatMap(pusher.push(_, ctx.metadata ++ metadata.getOrElse(Map.empty), delay)).as(none)
+            }
+          }
+          .flattenOption
+      }
+
+  /**
+   * Processes the messages with the provided message handler.
+   * The messages are ack'ed or nack'ed based on the decision returned from the handler.
+   * The stream emits results or errors down-stream and does not fail on business logic errors,
+   * allowing you to build error recovery logic.
+   *
+   * Messages in a batch are processed in parallel but result is emitted in order the messages were received,
+   * with the exclusion of the messages that have been dropped.
+   */
+  final def processWithImmediateDecision[Res](
+    batchSize: Int,
+    waitingTime: FiniteDuration
+  )(handler: MessageHandler[F, T, Res, ImmediateDecision]
+  ): Stream[F, Either[Throwable, Res]] =
+    process[Res](batchSize, waitingTime, QueuePublisher.noop)((msg: Message[F, T]) =>
+      handler.handle(msg).widen[Decision[Res]])
 }

core/src/test/scala/com/commercetools/queue/SubscriberSuite.scala

@@ -20,6 +20,7 @@ import cats.collections.Heap
 import cats.effect.IO
 import cats.effect.std.AtomicCell
 import cats.effect.testkit.TestControl
+import cats.implicits.catsSyntaxOptionId
 import cats.syntax.either._
 import cats.syntax.traverse._
 import com.commercetools.queue.testing._
@@ -35,11 +36,11 @@ class SubscriberSuite extends CatsEffectSuite {
       .map { state =>
         val queue =
           new TestQueue[String](name = "test-queue", state = state, messageTTL = 15.minutes, lockTTL = 1.minute)
-        (queue, new TestQueueSubscriber(queue))
+        (queue, new TestQueueSubscriber(queue), new TestQueuePublisher(queue))
       }
       .toResource)
 
-  queueSub.test("Successful messages must be acked") { case (queue, subscriber) =>
+  queueSub.test("Successful messages must be acked") { case (queue, subscriber, _) =>
     TestControl
       .executeEmbed(for {
         // first populate the queue
@@ -70,7 +71,7 @@ class SubscriberSuite extends CatsEffectSuite {
   }
 
   queueSub.test("Messages must be unack'ed if processing fails and emit everything up to failure") {
-    case (queue, subscriber) =>
+    case (queue, subscriber, _) =>
       TestControl
         .executeEmbed(for {
           // first populate the queue
@@ -98,4 +99,133 @@ class SubscriberSuite extends CatsEffectSuite {
         }
   }
 
+  queueSub.test("Messages consumed and ok'ed or drop'ed should follow the decision") {
+    case (queue, subscriber, publisher) =>
+      TestControl
+        .executeEmbed(for {
+          // first populate the queue
+          messages <- List.range(0, 100).traverse { i =>
+            IO.sleep(10.millis) *> IO.realTimeInstant.map(TestMessage(i.toString, _))
+          }
+          _ <- queue.setAvailableMessages(messages)
+          result <- subscriber
+            .process[Int](batchSize = 5, waitingTime = 40.millis, publisher)((msg: Message[IO, String]) =>
+              if (msg.rawPayload.toInt % 2 == 0) IO.pure(Decision.Drop)
+              else IO.pure(Decision.Ok(1)))
+            .interruptAfter(3.seconds)
+            .compile
+            .foldMonoid
+        } yield result)
+        .flatMap { result =>
+          for {
+            _ <- assertIO(queue.getAvailableMessages, Nil)
+            _ = assertEquals(result, 50.asRight)
+          } yield ()
+        }
+  }
+
+  queueSub.test("Messages consumed and confirmed or dropped should follow the decision") {
+    case (queue, subscriber, publisher) =>
+      TestControl
+        .executeEmbed(for {
+          // first populate the queue
+          messages <- List.range(0, 100).traverse { i =>
+            IO.sleep(10.millis) *> IO.realTimeInstant.map(TestMessage(i.toString, _))
+          }
+          _ <- queue.setAvailableMessages(messages)
+          result <- subscriber
+            .process[Int](batchSize = 5, waitingTime = 40.millis, publisher)((msg: Message[IO, String]) =>
+              if (msg.rawPayload.toInt % 2 == 0) IO.pure(Decision.Ok(1))
+              else IO.pure(Decision.Drop))
+            .take(50)
+            .compile
+            .foldMonoid
+        } yield result)
+        .flatMap { result =>
+          for {
+            _ <- assertIO(queue.getAvailableMessages, Nil)
+            _ = assertEquals(result, 50.asRight)
+          } yield ()
+        }
+  }
+
+  queueSub.test("Messages consumed and requeued should follow the decision") { case (queue, subscriber, publisher) =>
+    TestControl
+      .executeEmbed(for {
+        // first populate the queue
+        messages <- List.range(0, 100).traverse { i =>
+          IO.sleep(10.millis) *> IO.realTimeInstant.map(TestMessage(i.toString, _))
+        }
+        _ <- queue.setAvailableMessages(messages)
+        opCounter <- AtomicCell[IO].of(0)
+        result <- subscriber
+          .process[Int](batchSize = 5, waitingTime = 40.millis, publisher)((msg: Message[IO, String]) =>
+            opCounter.update(_ + 1) >> {
+              // let's reenqueue at the first run, and then confirm
+              if (msg.metadata.contains("reenqueued")) IO.pure(Decision.Ok(1))
+              else IO.pure(Decision.Reenqueue(Map("reenqueued" -> "true").some, None))
+            })
+          .take(100)
+          .compile
+          .foldMonoid
+        totOpCount <- opCounter.get
+      } yield (result, totOpCount))
+      .flatMap { case (result, totOpCount) =>
+        for {
+          _ <- assertIO(queue.getAvailableMessages, Nil)
+          _ = assertEquals(totOpCount, 200)
+          _ = assertEquals(result, 100.asRight)
+        } yield ()
+      }
+  }
+
+  queueSub.test("Messages that are marked as failed and acked should follow the decision") {
+    case (queue, subscriber, publisher) =>
+      TestControl
+        .executeEmbed(for {
+          // first populate the queue
+          messages <- List.range(0, 100).traverse { i =>
+            IO.sleep(10.millis) *> IO.realTimeInstant.map(TestMessage(i.toString, _))
+          }
+          _ <- queue.setAvailableMessages(messages)
+          result <- subscriber
+            .process[Int](batchSize = 5, waitingTime = 40.millis, publisher)((msg: Message[IO, String]) =>
+              IO.pure(Decision.Fail(new Throwable(s"failed ${msg.rawPayload}"), ack = true)))
+            .take(100)
+            .collect { case Left(_) => 1 }
+            .compile
+            .foldMonoid
+        } yield result)
+        .flatMap { result =>
+          for {
+            _ <- assertIO(queue.getAvailableMessages, Nil)
+            _ = assertEquals(result, 100)
+          } yield ()
+        }
+  }
+
+  queueSub.test("Messages that are marked as failed and not acked should follow the decision") {
+    case (queue, subscriber, publisher) =>
+      TestControl
+        .executeEmbed(for {
+          // first populate the queue
+          messages <- List.range(0, 100).traverse { i =>
+            IO.sleep(10.millis) *> IO.realTimeInstant.map(TestMessage(i.toString, _))
+          }
+          _ <- queue.setAvailableMessages(messages)
+          result <- subscriber
+            .process[Int](batchSize = 5, waitingTime = 40.millis, publisher)((msg: Message[IO, String]) =>
+              IO.pure(Decision.Fail(new Throwable(s"failed ${msg.rawPayload}"), ack = false)))
+            .take(100)
+            .collect { case Left(_) => 1 }
+            .compile
+            .foldMonoid
+        } yield result)
+        .flatMap { result =>
+          for {
+            _ <- assertIO(queue.getAvailableMessages.map(_.size), 100)
+            _ = assertEquals(result, 100)
+          } yield ()
+        }
+  }
 }

core/src/test/scala/com/commercetools/queue/testing/LockedTestMessage.scala

@@ -40,7 +40,7 @@ final case class LockedTestMessage[T](
 
   override def enqueuedAt: Instant = msg.enqueuedAt
 
-  override val metadata: Map[String, String] = Map.empty
+  override val metadata: Map[String, String] = msg.metadata
 
   override def ack(): IO[Unit] =
     // done, just delete it

core/src/test/scala/com/commercetools/queue/testing/TestMessage.scala

@@ -20,7 +20,7 @@ import cats.Order
 
 import java.time.Instant
 
-final case class TestMessage[T](payload: T, enqueuedAt: Instant)
+final case class TestMessage[T](payload: T, enqueuedAt: Instant, metadata: Map[String, String] = Map.empty)
 
 object TestMessage {
 

core/src/test/scala/com/commercetools/queue/testing/TestQueue.scala

@@ -126,10 +126,14 @@ class TestQueue[T](
         state <- update(state)
       } yield delay match {
         case None =>
-          state.copy(available = state.available.addAll(messages.map(x => TestMessage(x._1, now))))
+          state.copy(available = state.available.addAll(messages.map { case (payload, metadata) =>
+            TestMessage(payload, now, metadata)
+          }))
         case Some(delay) =>
           val delayed = now.plusMillis(delay.toMillis)
-          state.copy(delayed = messages.map(x => TestMessage(x._1, delayed)) reverse_::: state.delayed)
+          state.copy(delayed = messages.map { case (payload, metadata) =>
+            TestMessage(payload, delayed, metadata)
+          } reverse_::: state.delayed)
       }
     }