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

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.relayr" %% "rabbitmq-scala-client" % "$latestVersion"

To find the latest version please visit the project's page on search.maven.org.

Basic usage

Build connections with io.relayr.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