JarCasting
rabbitmq-scala-client
Wrapper around the rabbitmq-java-client for better scala usage. Much of this is based on amqp-client by sstone. The main reason for the rewrite is to not require our clients to use akka, to be easier to configure and to implement event hooks to enable statistics gathering.
Updates
- 0.1.6 - New methods to add consumers which accept an Envelope (encapsulating a message with the exchange and routing key it was sent to)
- 0.1.8 - Security & reliability fix - fix non-atomic update of RPC client call counter
- 0.2.0 - Compile for Scala 2.12
Features
- Sending and receiving of AMQP messages
- Support for Lyra reconnection strategies in the event of connection / channel / consumer failures
- Logging of dropped or returned messages, connection failures and reconnect attempts
- An implementation of the RPC pattern over AMQP
Download and inclusion on your project
The artifact is published to Maven Central. To add it to your build, add the following to your build.sbt:
libraryDependencies += "io.leonard" %% "amqp-scala" % "$latestVersion"
To find the latest version please visit the project's page on search.maven.org.
Basic usage
Build connections with io.leonard.amqp.ConnectionHolder.builder
Create a connection:
val connection = ConnectionHolder.builder("amqps://guest:password@host:port")
.eventHooks(EventHooks(eventListener))
.reconnectionStrategy(ReconnectionStrategy.JavaClientFixedReconnectDelay(1 second))
.build()
Create a channel:
val channel = connection.newChannel()
Create an RPC server listening on queue "queue.name", expecting a String and echoing it back:
def rpcHandler(request: Message): Future[Message] = request match {
case Message.String(string) => Future(Message.String(string))
}
val queue = QueueDeclare(Some("queue.name"))
val rpcServerCloser = channel.rpcServer(queue, AckOnHandled)(rpcHandler)
Create an RPC client method which sends requests to the queue "queue.name" with a response timeout of 10 seconds :
val rpcClient = RPCClient(channel)
val rpcMethod = rpcClient.newMethod(Exchange.Default.route("queue.name"), 10 second)
Create a consumer on "queue.name" printing out strings sent to it:
def consumer(request: Message): Unit = request match {
case Message.String(string) => println(string)
}
val queue = QueueDeclare(Some("queue.name"))
channel.addConsumer(queue, consumer)
Send a message to "queue.name":
channel.send(Exchange.Default.route("queue.name"), Message.String("message")
Contribution
Pull Requests welcome as well as Github issues