Merge pull request #972 from SheliakLyr/autoTopologyRecovery

Added automaticTopologyRecovery to Fs2RabbitConfig
profunktor · Sep 4, 2024 · 11024d0 · 11024d0
2 parents d41a64b + c611b6a
commit 11024d0
Show file tree

Hide file tree

Showing 6 changed files with 9 additions and 2 deletions.
diff --git a/core/src/main/scala/dev/profunktor/fs2rabbit/algebra/Connection.scala b/core/src/main/scala/dev/profunktor/fs2rabbit/algebra/Connection.scala
@@ -133,6 +133,7 @@ object ConnectionResource {
         factory.setConnectionTimeout(conf.connectionTimeout.toMillis.toInt)
         factory.setRequestedHeartbeat(conf.requestedHeartbeat.toSeconds.toInt)
         factory.setAutomaticRecoveryEnabled(conf.automaticRecovery)
+        factory.setTopologyRecoveryEnabled(conf.automaticTopologyRecovery)
         if (conf.ssl) sslCtx.fold(factory.useSslProtocol())(factory.useSslProtocol)
         factory.setSaslConfig(saslConf)
         conf.username.foreach(factory.setUsername)

diff --git a/core/src/main/scala/dev/profunktor/fs2rabbit/config/Fs2RabbitConfig.scala b/core/src/main/scala/dev/profunktor/fs2rabbit/config/Fs2RabbitConfig.scala
@@ -40,6 +40,7 @@ case class Fs2RabbitConfig(
     internalQueueSize: Option[Int],
     requestedHeartbeat: FiniteDuration,
     automaticRecovery: Boolean,
+    automaticTopologyRecovery: Boolean,
     clientProvidedConnectionName: Option[String]
 )
 
@@ -57,6 +58,7 @@ object Fs2RabbitConfig {
       internalQueueSize: Option[Int],
       requestedHeartbeat: FiniteDuration = FiniteDuration(ConnectionFactory.DEFAULT_HEARTBEAT, TimeUnit.SECONDS),
       automaticRecovery: Boolean = true,
+      automaticTopologyRecovery: Boolean = true,
       clientProvidedConnectionName: Option[String] = None
   ): Fs2RabbitConfig = Fs2RabbitConfig(
     nodes = NonEmptyList.one(Fs2RabbitNodeConfig(host, port)),
@@ -70,6 +72,7 @@ object Fs2RabbitConfig {
     internalQueueSize = internalQueueSize,
     requestedHeartbeat = requestedHeartbeat,
     automaticRecovery = automaticRecovery,
-    clientProvidedConnectionName = clientProvidedConnectionName
+    clientProvidedConnectionName = clientProvidedConnectionName,
+    automaticTopologyRecovery = automaticTopologyRecovery
   )
 }
diff --git a/examples/src/main/scala/dev/profunktor/fs2rabbit/examples/DropwizardMetricsDemo.scala b/examples/src/main/scala/dev/profunktor/fs2rabbit/examples/DropwizardMetricsDemo.scala
@@ -48,6 +48,7 @@ object DropwizardMetricsDemo extends IOApp.Simple {
     internalQueueSize = Some(500),
     requestedHeartbeat = 60.seconds,
     automaticRecovery = true,
+    automaticTopologyRecovery = true,
     clientProvidedConnectionName = Some("app:drop-wizard-metrics-demo")
   )
 

diff --git a/examples/src/main/scala/dev/profunktor/fs2rabbit/examples/IOAckerConsumer.scala b/examples/src/main/scala/dev/profunktor/fs2rabbit/examples/IOAckerConsumer.scala
@@ -38,6 +38,7 @@ object IOAckerConsumer extends IOApp.Simple {
     internalQueueSize = Some(500),
     requestedHeartbeat = 60.seconds,
     automaticRecovery = true,
+    automaticTopologyRecovery = true,
     clientProvidedConnectionName = Some("app:io-acker-consumer")
   )
 

diff --git a/examples/src/main/scala/dev/profunktor/fs2rabbit/examples/RPCDemo.scala b/examples/src/main/scala/dev/profunktor/fs2rabbit/examples/RPCDemo.scala
@@ -50,6 +50,7 @@ object RPCDemo extends IOApp.Simple {
     internalQueueSize = Some(500),
     requestedHeartbeat = 60.seconds,
     automaticRecovery = true,
+    automaticTopologyRecovery = true,
     clientProvidedConnectionName = Some("app:rpc-demo")
   )
 

diff --git a/tests/src/test/scala/dev/profunktor/fs2rabbit/interpreter/Fs2RabbitSpec.scala b/tests/src/test/scala/dev/profunktor/fs2rabbit/interpreter/Fs2RabbitSpec.scala
@@ -69,7 +69,7 @@ trait Fs2RabbitSpec { self: BaseSpec =>
     createConnectionChannel.use { implicit channel =>
       randomQueueData
         .flatMap { case (q, x, _) =>
-          declareQueue(DeclarationQueueConfig(q, Durable, Exclusive, AutoDelete, Map.empty)) *>
+          declareQueue(DeclarationQueueConfig(q, Durable, Exclusive, AutoDelete, Map.empty, None)) *>
             declareExchange(x, ExchangeType.Topic)
         }
         .as(emptyAssertion)