make timey config options finite durations

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

@@ -42,8 +42,8 @@ object ConnectionResource {
         factory.setHost(firstNode.host)
         factory.setPort(firstNode.port)
         factory.setVirtualHost(conf.virtualHost)
-        factory.setConnectionTimeout(conf.connectionTimeout)
-        factory.setRequestedHeartbeat(conf.requestedHeartbeat)
+        factory.setConnectionTimeout(conf.connectionTimeout.toMillis.toInt)
+        factory.setRequestedHeartbeat(conf.requestedHeartbeat.toSeconds.toInt)
         factory.setAutomaticRecoveryEnabled(conf.automaticRecovery)
         if (conf.ssl) sslCtx.fold(factory.useSslProtocol())(factory.useSslProtocol)
         factory.setSaslConfig(saslConf)

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

@@ -16,9 +16,13 @@
 
 package dev.profunktor.fs2rabbit.config
 
+import java.util.concurrent.TimeUnit
+
 import cats.data.NonEmptyList
 import com.rabbitmq.client.ConnectionFactory
 
+import scala.concurrent.duration.FiniteDuration
+
 case class Fs2RabbitNodeConfig(
     host: String,
     port: Int
@@ -27,14 +31,14 @@ case class Fs2RabbitNodeConfig(
 case class Fs2RabbitConfig(
     nodes: NonEmptyList[Fs2RabbitNodeConfig],
     virtualHost: String,
-    connectionTimeout: Int,
+    connectionTimeout: FiniteDuration,
     ssl: Boolean,
     username: Option[String],
     password: Option[String],
     requeueOnNack: Boolean,
     requeueOnReject: Boolean,
     internalQueueSize: Option[Int],
-    requestedHeartbeat: Int,
+    requestedHeartbeat: FiniteDuration,
     automaticRecovery: Boolean
 )
 
@@ -43,14 +47,14 @@ object Fs2RabbitConfig {
       host: String,
       port: Int,
       virtualHost: String,
-      connectionTimeout: Int,
+      connectionTimeout: FiniteDuration,
       ssl: Boolean,
       username: Option[String],
       password: Option[String],
       requeueOnNack: Boolean,
       requeueOnReject: Boolean,
       internalQueueSize: Option[Int],
-      requestedHeartbeat: Int = ConnectionFactory.DEFAULT_HEARTBEAT,
+      requestedHeartbeat: FiniteDuration = FiniteDuration(ConnectionFactory.DEFAULT_HEARTBEAT, TimeUnit.SECONDS),
       automaticRecovery: Boolean = true
   ): Fs2RabbitConfig = Fs2RabbitConfig(
     nodes = NonEmptyList.one(Fs2RabbitNodeConfig(host, port)),

examples/src/main/scala/dev/profunktor/fs2rabbit/examples/DropwizardMetricsDemo.scala

@@ -42,11 +42,11 @@ object DropwizardMetricsDemo extends IOApp.Simple {
     username = Some("guest"),
     password = Some("guest"),
     ssl = false,
-    connectionTimeout = 3,
+    connectionTimeout = 3.seconds,
     requeueOnNack = false,
     requeueOnReject = false,
     internalQueueSize = Some(500),
-    requestedHeartbeat = 60,
+    requestedHeartbeat = 60.seconds,
     automaticRecovery = true
   )
 

examples/src/main/scala/dev/profunktor/fs2rabbit/examples/IOAckerConsumer.scala

@@ -22,6 +22,8 @@ import dev.profunktor.fs2rabbit.config.{Fs2RabbitConfig, Fs2RabbitNodeConfig}
 import dev.profunktor.fs2rabbit.interpreter.RabbitClient
 import dev.profunktor.fs2rabbit.resiliency.ResilientStream
 
+import scala.concurrent.duration.DurationInt
+
 object IOAckerConsumer extends IOApp.Simple {
 
   private val config: Fs2RabbitConfig = Fs2RabbitConfig(
@@ -30,11 +32,11 @@ object IOAckerConsumer extends IOApp.Simple {
     username = Some("guest"),
     password = Some("guest"),
     ssl = false,
-    connectionTimeout = 3,
+    connectionTimeout = 3.seconds,
     requeueOnNack = false,
     requeueOnReject = false,
     internalQueueSize = Some(500),
-    requestedHeartbeat = 60,
+    requestedHeartbeat = 60.seconds,
     automaticRecovery = true
   )
 

examples/src/main/scala/dev/profunktor/fs2rabbit/examples/RPCDemo.scala

@@ -29,6 +29,8 @@ import dev.profunktor.fs2rabbit.interpreter.RabbitClient
 import dev.profunktor.fs2rabbit.model._
 import fs2.Stream
 
+import scala.concurrent.duration.DurationInt
+
 object RPCDemo extends IOApp.Simple {
 
   private val config: Fs2RabbitConfig = Fs2RabbitConfig(
@@ -42,11 +44,11 @@ object RPCDemo extends IOApp.Simple {
     username = Some("guest"),
     password = Some("guest"),
     ssl = false,
-    connectionTimeout = 3,
+    connectionTimeout = 3.seconds,
     requeueOnNack = false,
     requeueOnReject = false,
     internalQueueSize = Some(500),
-    requestedHeartbeat = 60,
+    requestedHeartbeat = 60.seconds,
     automaticRecovery = true
   )
 

site/docs/client.md

@@ -42,6 +42,8 @@ import cats.syntax.functor._
 import dev.profunktor.fs2rabbit.interpreter.RabbitClient
 import java.util.concurrent.Executors
 
+import scala.concurrent.duration._
+
 object Program {
   def foo[F[_]](client: RabbitClient[F]): F[Unit] = ???
 }
@@ -55,7 +57,7 @@ class Demo extends IOApp {
     password = Some("guest"),
     port = 5672,
     ssl = false,
-    connectionTimeout = 3,
+    connectionTimeout = 3.seconds,
     requeueOnNack = false,
     requeueOnReject = false,
     internalQueueSize = Some(500)

site/docs/config.md

@@ -12,6 +12,8 @@ The main `RabbitMQ` configuration should be defined as `Fs2RabbitConfig`. You ch
 import cats.data.NonEmptyList
 import dev.profunktor.fs2rabbit.config.{Fs2RabbitConfig, Fs2RabbitNodeConfig}
 
+import scala.concurrent.duration._
+
 val config = Fs2RabbitConfig(
   virtualHost = "/",
   nodes = NonEmptyList.one(
@@ -23,11 +25,11 @@ val config = Fs2RabbitConfig(
   username = Some("guest"),
   password = Some("guest"),
   ssl = false,
-  connectionTimeout = 3,
+  connectionTimeout = 3.seconds,
   requeueOnNack = false,
   requeueOnReject = false,
   internalQueueSize = Some(500),
-  requestedHeartbeat = 30,
+  requestedHeartbeat = 30.seconds,
   automaticRecovery = true
 )
 ```

site/docs/examples/client-metrics.md

@@ -104,11 +104,11 @@ object DropwizardMetricsDemo extends IOApp {
     username = Some("guest"),
     password = Some("guest"),
     ssl = false,
-    connectionTimeout = 3,
+    connectionTimeout = 3.seconds,
     requeueOnNack = false,
     requeueOnReject = false,
     internalQueueSize = Some(500),
-    requestedHeartbeat = 60,
+    requestedHeartbeat = 60.seconds,
     automaticRecovery = true
   )
 

site/docs/examples/sample-acker.md

@@ -96,6 +96,8 @@ import dev.profunktor.fs2rabbit.interpreter.RabbitClient
 import dev.profunktor.fs2rabbit.resiliency.ResilientStream
 import java.util.concurrent.Executors
 
+import scala.concurrent.duration.DurationInt
+
 object IOAckerConsumer extends IOApp {
 
   private val config: Fs2RabbitConfig = Fs2RabbitConfig(
@@ -109,11 +111,11 @@ object IOAckerConsumer extends IOApp {
     username = Some("guest"),
     password = Some("guest"),
     ssl = false,
-    connectionTimeout = 3,
+    connectionTimeout = 3.seconds,
     requeueOnNack = false,
     requeueOnReject = false,
     internalQueueSize = Some(500),
-    requestedHeartbeat = 60,
+    requestedHeartbeat = 60.seconds,
     automaticRecovery = true
   )
 

tests/src/test/scala/dev/profunktor/fs2rabbit/interpreter/RabbitSuite.scala

@@ -20,20 +20,22 @@ import cats.implicits._
 import dev.profunktor.fs2rabbit.BaseSpec
 import dev.profunktor.fs2rabbit.config.Fs2RabbitConfig
 
+import scala.concurrent.duration.DurationInt
+
 class RabbitSuite extends BaseSpec with Fs2RabbitSpec {
 
   override val config: Fs2RabbitConfig =
     Fs2RabbitConfig(
       host = "localhost",
       port = 5672,
       virtualHost = "/",
-      connectionTimeout = 30,
+      connectionTimeout = 30.seconds,
       ssl = false,
       username = "guest".some,
       password = "guest".some,
       requeueOnNack = false,
       requeueOnReject = false,
-      requestedHeartbeat = 60,
+      requestedHeartbeat = 60.seconds,
       internalQueueSize = 500.some
     )