Merge pull request #946 from geirolz/add-queue-type

Add QueueType model

Author: geirolz

core/src/main/scala/dev/profunktor/fs2rabbit/algebra/Declaration.scala

@@ -17,7 +17,7 @@
 package dev.profunktor.fs2rabbit.algebra
 
 import cats.effect.Sync
-import cats.syntax.functor._
+import cats.syntax.all._
 import dev.profunktor.fs2rabbit.arguments._
 import dev.profunktor.fs2rabbit.config.declaration.{DeclarationExchangeConfig, DeclarationQueueConfig}
 import dev.profunktor.fs2rabbit.effects.BoolValue.syntax._
@@ -63,26 +63,30 @@ object Declaration {
       }
 
     override def declareQueue(channel: AMQPChannel, config: DeclarationQueueConfig): F[Unit] =
-      Sync[F].blocking {
-        channel.value.queueDeclare(
-          config.queueName.value,
-          config.durable.isTrue,
-          config.exclusive.isTrue,
-          config.autoDelete.isTrue,
-          config.arguments
-        )
-      }.void
+      Sync[F].fromEither(config.validatedArguments).flatMap { args =>
+        Sync[F].blocking {
+          channel.value.queueDeclare(
+            config.queueName.value,
+            config.durable.isTrue,
+            config.exclusive.isTrue,
+            config.autoDelete.isTrue,
+            args
+          )
+        }.void
+      }
 
     override def declareQueueNoWait(channel: AMQPChannel, config: DeclarationQueueConfig): F[Unit] =
-      Sync[F].blocking {
-        channel.value.queueDeclareNoWait(
-          config.queueName.value,
-          config.durable.isTrue,
-          config.exclusive.isTrue,
-          config.autoDelete.isTrue,
-          config.arguments
-        )
-      }.void
+      Sync[F].fromEither(config.validatedArguments).flatMap { args =>
+        Sync[F].blocking {
+          channel.value.queueDeclareNoWait(
+            config.queueName.value,
+            config.durable.isTrue,
+            config.exclusive.isTrue,
+            config.autoDelete.isTrue,
+            args
+          )
+        }.void
+      }
 
     override def declareQueuePassive(channel: AMQPChannel, queueName: QueueName): F[Unit] =
       Sync[F].blocking {

core/src/main/scala/dev/profunktor/fs2rabbit/config/declaration.scala

@@ -17,7 +17,7 @@
 package dev.profunktor.fs2rabbit.config
 
 import dev.profunktor.fs2rabbit.arguments.Arguments
-import dev.profunktor.fs2rabbit.model.{ExchangeName, ExchangeType, QueueName}
+import dev.profunktor.fs2rabbit.model.{ExchangeName, ExchangeType, QueueName, QueueType}
 
 object declaration {
 
@@ -26,12 +26,44 @@ object declaration {
       durable: DurableCfg,
       exclusive: ExclusiveCfg,
       autoDelete: AutoDeleteCfg,
-      arguments: Arguments
-  )
+      arguments: Arguments,
+      queueType: Option[QueueType]
+  ) {
+
+    lazy val validatedArguments: Either[IllegalArgumentException, Arguments] =
+      queueType match {
+        case Some(_) if arguments.contains("x-queue-type") =>
+          Left(
+            new IllegalArgumentException(
+              "Queue type defined twice. It is set in the arguments and in the DeclarationQueueConfig."
+            )
+          )
+        case Some(queueType)                               =>
+          Right(arguments + ("x-queue-type" -> queueType.asString))
+        case None                                          =>
+          Right(arguments)
+      }
+  }
   object DeclarationQueueConfig {
 
     def default(queueName: QueueName): DeclarationQueueConfig =
-      DeclarationQueueConfig(queueName, NonDurable, NonExclusive, NonAutoDelete, Map.empty)
+      DeclarationQueueConfig(
+        queueName = queueName,
+        durable = NonDurable,
+        exclusive = NonExclusive,
+        autoDelete = NonAutoDelete,
+        arguments = Map.empty,
+        queueType = None
+      )
+
+    def classic(queueName: QueueName): DeclarationQueueConfig =
+      default(queueName).copy(queueType = Some(QueueType.Classic))
+
+    def quorum(queueName: QueueName): DeclarationQueueConfig =
+      default(queueName).copy(queueType = Some(QueueType.Quorum))
+
+    def stream(queueName: QueueName): DeclarationQueueConfig =
+      default(queueName).copy(queueType = Some(QueueType.Stream))
   }
 
   sealed trait DurableCfg extends Product with Serializable

core/src/main/scala/dev/profunktor/fs2rabbit/model.scala

@@ -93,8 +93,20 @@ object model {
         extends ExchangeType // for use with the plugin https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/
   }
 
-  sealed abstract class DeliveryMode(val value: Int) extends Product with Serializable
+  sealed trait QueueType extends Product with Serializable {
+    def asString: String = this match {
+      case QueueType.Classic => "classic"
+      case QueueType.Quorum  => "quorum"
+      case QueueType.Stream  => "stream"
+    }
+  }
+  object QueueType {
+    case object Classic extends QueueType
+    case object Quorum  extends QueueType
+    case object Stream  extends QueueType
+  }
 
+  sealed abstract class DeliveryMode(val value: Int) extends Product with Serializable
   object DeliveryMode {
     case object NonPersistent extends DeliveryMode(1)
     case object Persistent    extends DeliveryMode(2)