Merge pull request #165 from mturlo/master

exposed consumer.exclusive flag in consume() directive
SpinGo · Mar 2, 2019 · 9a681c4 · 9a681c4
2 parents 85283cf + f14aedc
commit 9a681c4
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 5 deletions.
diff --git a/core/src/main/scala/com/spingo/op_rabbit/Directives.scala b/core/src/main/scala/com/spingo/op_rabbit/Directives.scala
@@ -22,10 +22,11 @@ private [op_rabbit] case class BoundConsumerDefinition(
   recoveryStrategy: RecoveryStrategy,
   executionContext: ExecutionContext,
   consumerArgs: Seq[properties.Header],
-  consumerTagPrefix: Option[String])
-private [op_rabbit] case class BindingDirective(binding: QueueDefinition[Concreteness], args: Seq[properties.Header], consumerTagPrefix: Option[String]) {
+  consumerTagPrefix: Option[String],
+  exclusive: Boolean)
+private [op_rabbit] case class BindingDirective(binding: QueueDefinition[Concreteness], args: Seq[properties.Header], consumerTagPrefix: Option[String], exclusive: Boolean) {
   def apply(thunk: => Handler)(implicit errorReporting: RabbitErrorLogging, recoveryStrategy: RecoveryStrategy, executionContext: ExecutionContext) =
-    BoundConsumerDefinition(binding, handler = thunk, errorReporting, recoveryStrategy, executionContext, args, consumerTagPrefix)
+    BoundConsumerDefinition(binding, handler = thunk, errorReporting, recoveryStrategy, executionContext, args, consumerTagPrefix, exclusive)
 }
 private [op_rabbit] case class ChannelConfiguration(qos: Int)
 private [op_rabbit] case class BoundChannel(channelConfig: ChannelConfiguration, boundConsumer: BoundConsumerDefinition)
@@ -89,7 +90,8 @@ trait Directives {
   def consume(
     binding: QueueDefinition[Concreteness],
     args: Seq[properties.Header] = Seq(),
-    consumerTagPrefix: Option[String] = None) = BindingDirective(binding, args, consumerTagPrefix)
+    consumerTagPrefix: Option[String] = None,
+    exclusive: Boolean = false) = BindingDirective(binding, args, consumerTagPrefix, exclusive)
 
   /**
     Provides values for the [[consume]] directive.

diff --git a/core/src/main/scala/com/spingo/op_rabbit/impl/AsyncAckingRabbitConsumer.scala b/core/src/main/scala/com/spingo/op_rabbit/impl/AsyncAckingRabbitConsumer.scala
@@ -125,7 +125,7 @@ private [op_rabbit] class AsyncAckingRabbitConsumer[T](
       false,
       subscription.consumerTagPrefix.fold("")(prefix => s"$prefix-${ConsumerId()}"),
       false,
-      false,
+      subscription.exclusive,
       properties.toJavaMap(subscription.consumerArgs),
       new DefaultConsumer(channel) {
         override def handleDelivery(