From d3a5dc26983f3e89c278cc4cf3858e3db1af52fb Mon Sep 17 00:00:00 2001 From: "liviu.ungureanu" Date: Fri, 21 Feb 2020 11:29:03 +0000 Subject: [PATCH 1/2] expose to the client the akka-rabbitmq ConnectionActor's connectionSetup callback. --- .../scala/com/spingo/op_rabbit/RabbitControl.scala | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/com/spingo/op_rabbit/RabbitControl.scala b/core/src/main/scala/com/spingo/op_rabbit/RabbitControl.scala index ba96f80..60e910d 100644 --- a/core/src/main/scala/com/spingo/op_rabbit/RabbitControl.scala +++ b/core/src/main/scala/com/spingo/op_rabbit/RabbitControl.scala @@ -3,7 +3,8 @@ package com.spingo.op_rabbit import akka.actor.SupervisorStrategy._ import akka.actor._ import akka.util.Timeout -import com.newmotion.akka.rabbitmq.{ ConnectionActor, CreateChannel, ChannelActor, ChannelCreated, ChannelMessage } +import com.newmotion.akka.rabbitmq.{ChannelActor, ChannelCreated, ChannelMessage, Connection, ConnectionActor, CreateChannel} + import scala.concurrent.Promise import scala.concurrent.duration._ @@ -67,9 +68,14 @@ private [op_rabbit] class Sequence extends Iterator[Int] { * `akka-rabbitmq` ConnectionActor * - [[Subscription]] - Activate the given subscription; responds with a [[SubscriptionRef]] */ -class RabbitControl(connection: Either[ConnectionParams, ActorRef]) extends Actor with ActorLogging with Stash { +class RabbitControl(connection: Either[ConnectionParams, ActorRef], + setupConnectionCallback: (Connection, ActorRef) => Any = (_, _) => ()) extends Actor with ActorLogging with Stash { def this() = this(Left(ConnectionParams.fromConfig())) def this(connectionParams: ConnectionParams) = this(Left(connectionParams)) + def this(connectionParams: ConnectionParams, setupConnectionCallback: (Connection, ActorRef) => Any) = { + this(Left(connectionParams), setupConnectionCallback) + } + def this(actorRef: ActorRef) = this(Right(actorRef)) val sequence = new Sequence @@ -93,7 +99,7 @@ class RabbitControl(connection: Either[ConnectionParams, ActorRef]) extends Acto val connectionFactory = new ClusterConnectionFactory connectionParams.applyTo(connectionFactory) context.actorOf( - ConnectionActor.props(connectionFactory), + ConnectionActor.props(connectionFactory, setupConnection = setupConnectionCallback), name = CONNECTION_ACTOR_NAME) case Right(actorRef) => actorRef From cc12f0936be1a33495132b6a7c5062a47e6238ec Mon Sep 17 00:00:00 2001 From: Ungureanu Date: Mon, 25 May 2020 12:30:09 +0300 Subject: [PATCH 2/2] expose channelRpcTimeout --- .../main/scala/com/spingo/op_rabbit/ConnectionParams.scala | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/com/spingo/op_rabbit/ConnectionParams.scala b/core/src/main/scala/com/spingo/op_rabbit/ConnectionParams.scala index 59650c7..2836380 100644 --- a/core/src/main/scala/com/spingo/op_rabbit/ConnectionParams.scala +++ b/core/src/main/scala/com/spingo/op_rabbit/ConnectionParams.scala @@ -42,7 +42,8 @@ case class ConnectionParams( sharedExecutor: Option[java.util.concurrent.ExecutorService] = None, shutdownTimeout: Int = ConnectionFactory.DEFAULT_SHUTDOWN_TIMEOUT, socketFactory: SocketFactory = SocketFactory.getDefault, - sslContextOpt: Option[SSLContext] = None + sslContextOpt: Option[SSLContext] = None, + channelRpcTimeout: Int = ConnectionFactory.DEFAULT_CHANNEL_RPC_TIMEOUT ) { // TODO - eliminate ClusterConnectionFactory after switching to use RabbitMQ's topology recovery features. protected [op_rabbit] def applyTo(factory: ClusterConnectionFactory): Unit = { @@ -71,6 +72,8 @@ case class ConnectionParams( factory.useSslProtocol() } } + + factory.setChannelRpcTimeout(channelRpcTimeout) } }