Merge pull request #215 from changlinli/fix-212

Add option to pass in SASL config
profunktor · Jul 2, 2019 · 6fca3de · 6fca3de
2 parents a95fd55 + 617bf46
commit 6fca3de
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 4 deletions.
diff --git a/core/src/main/scala/dev/profunktor/fs2rabbit/interpreter/ConnectionEffect.scala b/core/src/main/scala/dev/profunktor/fs2rabbit/interpreter/ConnectionEffect.scala
@@ -19,7 +19,7 @@ package dev.profunktor.fs2rabbit.interpreter
 import cats.data.NonEmptyList
 import cats.effect.{Resource, Sync}
 import cats.implicits._
-import com.rabbitmq.client.{Address, ConnectionFactory}
+import com.rabbitmq.client.{Address, ConnectionFactory, SaslConfig}
 import dev.profunktor.fs2rabbit.algebra.Connection
 import dev.profunktor.fs2rabbit.config.Fs2RabbitConfig
 import dev.profunktor.fs2rabbit.effects.Log
@@ -62,7 +62,8 @@ object ConnectionEffect {
 
   private[fs2rabbit] def mkConnectionFactory[F[_]: Sync](
       config: Fs2RabbitConfig,
-      sslContext: Option[SSLContext]
+      sslContext: Option[SSLContext],
+      saslConfig: SaslConfig
   ): F[(ConnectionFactory, NonEmptyList[Address])] =
     Sync[F].delay {
       val factory   = new ConnectionFactory()
@@ -75,6 +76,7 @@ object ConnectionEffect {
       if (config.ssl) {
         sslContext.fold(factory.useSslProtocol())(factory.useSslProtocol)
       }
+      factory.setSaslConfig(saslConfig)
       config.username.foreach(factory.setUsername)
       config.password.foreach(factory.setPassword)
       val addresses = config.nodes.map(node => new Address(node.host, node.port))

diff --git a/core/src/main/scala/dev/profunktor/fs2rabbit/interpreter/Fs2Rabbit.scala b/core/src/main/scala/dev/profunktor/fs2rabbit/interpreter/Fs2Rabbit.scala
@@ -18,6 +18,7 @@ package dev.profunktor.fs2rabbit.interpreter
 
 import cats.effect.{Concurrent, ConcurrentEffect, Resource}
 import cats.syntax.functor._
+import com.rabbitmq.client.{DefaultSaslConfig, SaslConfig}
 import dev.profunktor.fs2rabbit.algebra._
 import dev.profunktor.fs2rabbit.config.Fs2RabbitConfig
 import dev.profunktor.fs2rabbit.config.declaration.{DeclarationExchangeConfig, DeclarationQueueConfig}
@@ -31,9 +32,12 @@ import javax.net.ssl.SSLContext
 object Fs2Rabbit {
   def apply[F[_]: ConcurrentEffect](
       config: Fs2RabbitConfig,
-      sslContext: Option[SSLContext] = None
+      sslContext: Option[SSLContext] = None,
+      // Unlike SSLContext, SaslConfig is not optional because it is always set
+      // by the underlying Java library, even if the user doesn't set it.
+      saslConfig: SaslConfig = DefaultSaslConfig.PLAIN
   ): F[Fs2Rabbit[F]] =
-    ConnectionEffect.mkConnectionFactory[F](config, sslContext).map {
+    ConnectionEffect.mkConnectionFactory[F](config, sslContext, saslConfig).map {
       case (factory, addresses) =>
         val amqpClient = new AmqpClientEffect[F]
         val conn       = new ConnectionEffect[F](factory, addresses)