profunktor
diff --git a/‎core/src/main/scala/dev/profunktor/fs2rabbit/algebra/Consume.scala‎
Lines changed: 6 additions & 1 deletion b/‎core/src/main/scala/dev/profunktor/fs2rabbit/algebra/Consume.scala‎
Lines changed: 6 additions & 1 deletion
diff --git a/‎docker-compose.yml‎
Lines changed: 2 additions & 0 deletions b/‎docker-compose.yml‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎rabbit-test-config/advanced.config‎
Lines changed: 6 additions & 0 deletions b/‎rabbit-test-config/advanced.config‎
Lines changed: 6 additions & 0 deletions
diff --git a/‎rabbit-test-config/rabbitmq.conf‎
Lines changed: 1 addition & 0 deletions b/‎rabbit-test-config/rabbitmq.conf‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎tests/src/test/scala/dev/profunktor/fs2rabbit/interpreter/Fs2RabbitSpec.scala‎
Lines changed: 22 additions & 0 deletions b/‎tests/src/test/scala/dev/profunktor/fs2rabbit/interpreter/Fs2RabbitSpec.scala‎
Lines changed: 22 additions & 0 deletions
@@ -21,7 +21,7 @@ import cats.effect.std.Dispatcher
 import cats.syntax.flatMap._
 import cats.syntax.functor._
 import cats.{Applicative, Functor}
-import com.rabbitmq.client.{AMQP, Consumer, DefaultConsumer, Envelope}
+import com.rabbitmq.client.{AMQP, Consumer, DefaultConsumer, Envelope, ShutdownSignalException}
 import dev.profunktor.fs2rabbit.arguments.{Arguments, _}
 import dev.profunktor.fs2rabbit.model._
 
@@ -106,6 +106,11 @@ object Consume {
                 }
             }
           }
+
+          override def handleShutdownSignal(consumerTag: String, sig: ShutdownSignalException): Unit =
+            if (!sig.isInitiatedByApplication) {
+              internals.queue.foreach(q => dispatcher.unsafeRunAndForget(q.offer(Left(sig))))
+            }
         }
       }
 
 
@@ -5,3 +5,5 @@ RabbitMQ:
     - "5672:5672"
   environment:
     - DEBUG=false
+  volumes:
+    - ./rabbit-test-config/:/etc/rabbitmq/
@@ -0,0 +1,6 @@
+[
+    {rabbit, [
+        {channel_tick_interval, 500},
+        {loopback_users, []}
+    ]}
+].
@@ -0,0 +1 @@
+consumer_timeout = 1000
@@ -711,6 +711,28 @@ trait Fs2RabbitSpec { self: BaseSpec =>
       }
   }
 
+  it should "shutdown the stream when the server closes the channel" in withRabbit { interpreter =>
+    import interpreter._
+    val msg = "will-not-be-acked"
+    createConnectionChannel.use { implicit channel =>
+      for {
+        qxrk      <- randomQueueData
+        (q, x, rk) = qxrk
+        _         <- declareExchange(x, ExchangeType.Topic)
+        _         <- declareQueue(DeclarationQueueConfig.default(q))
+        _         <- bindQueue(q, x, rk, QueueBindingArgs(Map.empty))
+        publisher <- createPublisher[String](x, rk)
+        _         <- publisher(msg)
+        stream    <- createAckerConsumer(q).map(_._2)
+        results   <- stream.attempt.compile.toList.timeoutAndForget(Duration(10, "s"))
+      } yield {
+        results.size shouldEqual 2
+        results.head.map(_.payload) shouldEqual Right(msg)
+        results.last.isLeft shouldEqual true
+      }
+    }
+  }
+
   it should "preserve order of published messages" in withRabbit { interpreter =>
     import dev.profunktor.fs2rabbit.effects.{EnvelopeDecoder, MessageEncoder}
     import interpreter._
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ import cats.effect.std.Dispatcher`
`21`	`21`	`import cats.syntax.flatMap._`
`22`	`22`	`import cats.syntax.functor._`
`23`	`23`	`import cats.{Applicative, Functor}`
`24`		`-import com.rabbitmq.client.{AMQP, Consumer, DefaultConsumer, Envelope}`
	`24`	`+import com.rabbitmq.client.{AMQP, Consumer, DefaultConsumer, Envelope, ShutdownSignalException}`
`25`	`25`	`import dev.profunktor.fs2rabbit.arguments.{Arguments, _}`
`26`	`26`	`import dev.profunktor.fs2rabbit.model._`
`27`	`27`
`@@ -106,6 +106,11 @@ object Consume {`
`106`	`106`	`}`
`107`	`107`	`}`
`108`	`108`	`}`
	`109`	`+`
	`110`	`+ override def handleShutdownSignal(consumerTag: String, sig: ShutdownSignalException): Unit =`
	`111`	`+ if (!sig.isInitiatedByApplication) {`
	`112`	`+ internals.queue.foreach(q => dispatcher.unsafeRunAndForget(q.offer(Left(sig))))`
	`113`	`+ }`
`109`	`114`	`}`
`110`	`115`	`}`
`111`	`116`