Development focus has switched to pekko-rabbitmq.
This small library allows you to use RabbitMQ client via Akka Actors.
This is a fork of NewMotion/akka-rabbitmq but adds Scala 3 support.
It gives you two actors ConnectionActor
and ChannelActor
.
- handles connection failures and notifies children
- keep trying to reconnect if connection lost
- provides children with new channels when needed
- may store messages in memory if channel lost
- send stored messages as soon as new channel received
- retrieve new channel if current is broken
Please note that while this library transparently reconnects when a connection fails, it cannot guarantee that no messages will be lost. If you want to make sure every message is delivered, you have to use acknowledgements and confirms. This is documented in the RabbitMQ Reliability Guide. An example program using confirms can be found in this project under ConfirmsExample.scala.
libraryDependencies += "com.github.pjfanning" %% "akka-rabbitmq" % "6.1.0"
<dependency>
<groupId>com.github.pjfanning</groupId>
<artifactId>akka-rabbitmq_{2.12/2.13/3}</artifactId>
<version>6.1.0</version>
</dependency>
Before start, you need to add import statement
import com.newmotion.akka.rabbitmq._
Default approach:
val factory = new ConnectionFactory()
val connection: Connection = factory.newConnection()
Actor style:
val factory = new ConnectionFactory()
val connectionActor: ActorRef = system.actorOf(ConnectionActor.props(factory))
Let's name it:
system.actorOf(ConnectionActor.props(factory), "my-connection")
How often will it reconnect?
import concurrent.duration._
system.actorOf(ConnectionActor.props(factory, reconnectionDelay = 10.seconds), "my-connection")
That's plain option:
val channel: Channel = connection.createChannel()
But we can do better. Asynchronously:
connectionActor ! CreateChannel(ChannelActor.props())
Synchronously:
val channelActor: ActorRef = connectionActor.createChannel(ChannelActor.props())
Maybe give it a name:
connectionActor.createChannel(ChannelActor.props(), Some("my-channel"))
What's about custom actor:
connectionActor.createChannel(Props(new Actor {
def receive = {
case channel: Channel =>
}
}))
channel.queueDeclare("queue_name", false, false, false, null)
Actor style:
// this function will be called each time new channel received
def setupChannel(channel: Channel, self: ActorRef) = {
channel.queueDeclare("queue_name", false, false, false, null)
}
val channelActor: ActorRef = connectionActor.createChannel(ChannelActor.props(setupChannel))
channel.basicPublish("", "queue_name", null, "Hello world".getBytes)
Using our channelActor
:
def publish(channel: Channel) = {
channel.basicPublish("", "queue_name", null, "Hello world".getBytes)
}
channelActor ! ChannelMessage(publish)
But I don't want to lose messages when connection is lost:
channelActor ! ChannelMessage(publish, dropIfNoChannel = false)
channel.close()
VS
system stop channelActor
connection.close()
VS
system stop connectionActor // will close all channels associated with this connection
You can shutdown ActorSystem
, this will close all connections as well as channels:
system.shutdown()
Here is RabbitMQ Publish/Subscribe in actors style
import akka.actor.ActorSystem
object PublishSubscribe extends App {
implicit val system: ActorSystem = ActorSystem()
val factory = new ConnectionFactory()
val connection = system.actorOf(ConnectionActor.props(factory), "akka-rabbitmq")
val exchange = "amq.fanout"
def setupPublisher(channel: Channel, self: ActorRef) = {
val queue = channel.queueDeclare().getQueue
channel.queueBind(queue, exchange, "")
}
connection ! CreateChannel(ChannelActor.props(setupPublisher), Some("publisher"))
def setupSubscriber(channel: Channel, self: ActorRef) = {
val queue = channel.queueDeclare().getQueue
channel.queueBind(queue, exchange, "")
val consumer = new DefaultConsumer(channel) {
override def handleDelivery(consumerTag: String, envelope: Envelope, properties: BasicProperties, body: Array[Byte]): Unit = {
println("received: " + fromBytes(body))
}
}
channel.basicConsume(queue, true, consumer)
}
connection ! CreateChannel(ChannelActor.props(setupSubscriber), Some("subscriber"))
Future {
def loop(n: Long) = {
val publisher = system.actorSelection("/user/akka-rabbitmq/publisher")
def publish(channel: Channel) = {
channel.basicPublish(exchange, "", null, toBytes(n))
}
publisher ! ChannelMessage(publish, dropIfNoChannel = false)
Thread.sleep(1000)
loop(n + 1)
}
loop(0)
}
def fromBytes(x: Array[Byte]) = new String(x, "UTF-8")
def toBytes(x: Long) = x.toString.getBytes("UTF-8")
}
Tests can be run against a RabbitMQ server on the local machine using a Docker container with the following command. The RabbitMQ console can be accessible also with http://localhost:8080 using the login and password of guest and guest.
docker run -d -p:5672:5672 rabbitmq:3
Akka-RabbitMQ is a low-level library, and leaves it to the coder to manually wire consumers, serialize messages, etc. If you'd like a higher-level abstraction library, look at pjfanning/op-rabbit (which uses this library).