Network recovery strategy (handlers) - linear, exponential (#23)

* Network recovery strategy (handlers) - linear, exponential
* Readme update

Author: jendakol

Migration-6_1-7.md

@@ -1,6 +1,10 @@
 ## Migration from 6.1.x to 7.0.x
 
-Additional declarations: `bindQueue` now has more consistent API. The only change is `bindArguments` were renamed to `arguments`.
+Common changes:
+
+1. Additional declarations: `bindQueue` now has more consistent API. The only change is `bindArguments` were renamed to `arguments`.
+1. You are able to specify network recovery strategy.
+1. You are able to specify timeout log level.
 
 Changes in Scala API:
 

README.md

@@ -28,7 +28,7 @@ some optional functionality:
 ## Migration
 
 There is a [migration guide](Migration-5-6.md) between versions 5 and 6.0.x.  
-There is a [migration guide](Migration-6-6_1.md) between versions 6.0.x and 6.1.x.
+There is a [migration guide](Migration-6-6_1.md) between versions 6.0.x and 6.1.x.  
 There is a [migration guide](Migration-6_1-7.md) between versions 6.1.x and 7.0.x.
 
 ## Usage
@@ -270,6 +270,11 @@ However there exists a workaround:
 1. Convert it to your `F[_]` by providing `cats.arrow.FunctionK[Task, A]` and `cats.arrow.FunctionK[A, Task]`
 
 ```scala
+import monix.eval.Task
+import scala.concurrent.Future
+import scala.concurrent.duration._
+import com.avast.clients.rabbitmq._
+
 implicit val fkToFuture: cats.arrow.FunctionK[Task, Future] = ???
 implicit val fkFromFuture: cats.arrow.FunctionK[Future, Task] = ???
 
@@ -369,6 +374,18 @@ See [full example](core/src/test/java/ExampleJava.java)
 ### Extras
 There is a module with some optional functionality called [extras](extras/README.md).
 
+### Network recovery
+The library offers configurable network recovery, with the functionality itself backed by RabbitMQ client's one (ready in 5+).  
+You can either disable the recovery or select (and configure one of following types):
+1. Linear  
+    The client will wait `initialDelay` for first recovery attempt and if it fails, will try it again each `period` until it succeeds.
+1. Exponential  
+    The client will wait `initialDelay` for first recovery attempt and if it fails, will try it again until it succeeds and prolong the
+    delay between each two attempts exponentially (based on `period`, `factor`, attempt number), up to `maxLength`.  
+    Example:  
+    For `initialDelay = 3s, period = 2s, factor = 2.0, maxLength = 1 minute`, produced delays will be 3, 2, 4, 8, 16, 32, 60 seconds
+    (and it will never go higher).
+
 ### DeliveryResult
 The consumers `readAction` returns `Future` of [`DeliveryResult`](api/src/main/scala/com/avast/clients/rabbitmq/api/DeliveryResult.scala). The `DeliveryResult` has 4 possible values
 (descriptions of usual use-cases):

core/src/main/resources/reference.conf

@@ -27,10 +27,24 @@ avastRabbitMQConnectionDefaults {
 
   networkRecovery {
     enabled = true
-    period = 5s
+    type = "linear" // exponential, linear
+
+    // merged with avastRabbitMQRecoveryLinearDefaults or avastRabbitMQRecoveryExponentialDefaults, see below
   }
 }
 
+avastRabbitMQRecoveryLinearDefaults {
+  initialDelay = 1s
+  period = 5s
+}
+
+avastRabbitMQRecoveryExponentialDefaults {
+  initialDelay = 1s
+  period = 5s
+  factor = 2.0
+  maxLength = 32s
+}
+
 avastRabbitMQConsumerDefaults {
   //  name = ""
 

core/src/main/scala/com/avast/clients/rabbitmq/DefaultRabbitMQProducer.scala

@@ -52,7 +52,7 @@ class DefaultRabbitMQProducer[F[_], A: ProductConverter](name: String,
 
   private def send(routingKey: String, body: Bytes, properties: MessageProperties): Task[Unit] = {
     Task {
-      logger.debug(s"Sending message with ${body.size()} B to exchange $exchangeName with routing key $routingKey and $properties")
+      logger.debug(s"Sending message with ${body.size()} B to exchange $exchangeName with routing key '$routingKey' and $properties")
 
       try {
         sendLock.synchronized {

core/src/main/scala/com/avast/clients/rabbitmq/RabbitMQConnection.scala

@@ -122,13 +122,41 @@ object RabbitMQConnection extends StrictLogging {
   }
 
   private[rabbitmq] final val RootConfigKey = "avastRabbitMQConnectionDefaults"
-
   private[rabbitmq] final val DefaultConfig = ConfigFactory.defaultReference().getConfig(RootConfigKey)
+  private[rabbitmq] final val RootConfigKeyRecoveryLinear = "avastRabbitMQRecoveryLinearDefaults"
+  private[rabbitmq] final val DefaultConfigRecoveryLinear = ConfigFactory.defaultReference().getConfig(RootConfigKeyRecoveryLinear)
+  private[rabbitmq] final val RootConfigKeyRecoveryExponential = "avastRabbitMQRecoveryExponentialDefaults"
+  private[rabbitmq] final val DefaultConfigRecoveryExponential =
+    ConfigFactory.defaultReference().getConfig(RootConfigKeyRecoveryExponential)
 
   private implicit final val JavaDurationReader: ValueReader[Duration] = (config: Config, path: String) => config.getDuration(path)
 
   private implicit final val JavaPathReader: ValueReader[Path] = (config: Config, path: String) => Paths.get(config.getString(path))
 
+  private implicit final val RecoveryDelayHandlerReader: ValueReader[RecoveryDelayHandler] = (config: Config, path: String) => {
+    val rdhConfig = config.getConfig(path.split('.').dropRight(1).mkString("."))
+
+    rdhConfig.getString("type").toLowerCase match {
+      case "linear" =>
+        val finalConfig = rdhConfig.withFallback(DefaultConfigRecoveryLinear)
+
+        RecoveryDelayHandlers.Linear(
+          delay = finalConfig.getDuration("initialDelay"),
+          period = finalConfig.getDuration("period")
+        )
+
+      case "exponential" =>
+        val finalConfig = rdhConfig.withFallback(DefaultConfigRecoveryExponential)
+
+        RecoveryDelayHandlers.Exponential(
+          delay = finalConfig.getDuration("initialDelay"),
+          period = finalConfig.getDuration("period"),
+          factor = finalConfig.getDouble("factor"),
+          maxLength = finalConfig.getDuration("maxLength"),
+        )
+    }
+  }
+
   /** Creates new instance of channel factory, using the passed configuration.
     *
     * @param providedConfig   The configuration.
@@ -210,11 +238,12 @@ object RabbitMQConnection extends StrictLogging {
     factory.setVirtualHost(virtualHost)
 
     factory.setTopologyRecoveryEnabled(topologyRecovery)
-    factory.setAutomaticRecoveryEnabled(true)
-    factory.setNetworkRecoveryInterval(networkRecovery.period.toMillis)
+    factory.setAutomaticRecoveryEnabled(networkRecovery.enabled)
     factory.setExceptionHandler(exceptionHandler)
     factory.setRequestedHeartbeat(heartBeatInterval.getSeconds.toInt)
 
+    if (networkRecovery.enabled) factory.setRecoveryDelayHandler(networkRecovery.handler)
+
     factory.setSharedExecutor(executor)
 
     if (credentials.enabled) {

core/src/main/scala/com/avast/clients/rabbitmq/RecoveryDelayHandlers.scala

@@ -0,0 +1,26 @@
+package com.avast.clients.rabbitmq
+import java.time.Duration
+
+import com.rabbitmq.client.RecoveryDelayHandler
+
+private[rabbitmq] object RecoveryDelayHandlers {
+  case class Linear(delay: Duration, period: Duration) extends RecoveryDelayHandler {
+    override def getDelay(recoveryAttempts: Int): Long = {
+      if (recoveryAttempts == 0) delay.toMillis else period.toMillis
+    }
+  }
+
+  case class Exponential(delay: Duration, period: Duration, factor: Double, maxLength: Duration) extends RecoveryDelayHandler {
+    private val maxMillis = maxLength.toMillis
+
+    override def getDelay(recoveryAttempts: Int): Long = {
+      if (recoveryAttempts == 0) delay.toMillis
+      else {
+        math.min(
+          maxMillis,
+          (period.toMillis * math.pow(factor, recoveryAttempts - 1)).toLong
+        )
+      }
+    }
+  }
+}

core/src/main/scala/com/avast/clients/rabbitmq/configuration.scala

@@ -4,6 +4,7 @@ import java.nio.file.Path
 import java.time.Duration
 
 import com.avast.clients.rabbitmq.api.DeliveryResult
+import com.rabbitmq.client.RecoveryDelayHandler
 import com.typesafe.config.Config
 import org.slf4j.event.Level
 
@@ -19,7 +20,7 @@ case class RabbitMQConnectionConfig(hosts: Array[String],
                                     credentials: Credentials,
                                     ssl: Ssl)
 
-case class NetworkRecovery(enabled: Boolean, period: Duration)
+case class NetworkRecovery(enabled: Boolean, handler: RecoveryDelayHandler)
 
 case class Credentials(enabled: Boolean, username: String, password: String)
 

core/src/test/scala/com/avast/clients/rabbitmq/RecoveryDelayHandlersTest.scala

@@ -0,0 +1,30 @@
+package com.avast.clients.rabbitmq
+import java.time.Duration
+
+import scala.util.Random
+
+class RecoveryDelayHandlersTest extends TestBase {
+  test("linear") {
+    val rdh = RecoveryDelayHandlers.Linear(Duration.ofMillis(10), Duration.ofMillis(42))
+
+    assertResult(10)(rdh.getDelay(0))
+
+    for (_ <- 1 to 200) {
+      assertResult(42)(rdh.getDelay(Random.nextInt() + 1))
+    }
+  }
+
+  test("exponential") {
+    val rdh = RecoveryDelayHandlers.Exponential(Duration.ofMillis(1), Duration.ofMillis(5), 2.0, Duration.ofMillis(42))
+
+    assertResult(1)(rdh.getDelay(0))
+
+    assertResult(5)(rdh.getDelay(1))
+    assertResult(10)(rdh.getDelay(2))
+    assertResult(20)(rdh.getDelay(3))
+    assertResult(40)(rdh.getDelay(4))
+    assertResult(42)(rdh.getDelay(5))
+    assertResult(42)(rdh.getDelay(6))
+    assertResult(42)(rdh.getDelay(7))
+  }
+}