Reactive Streams: AMQP

Reactive Streams driver for AMQP protocol. Powered by RabbitMQ library.

Available at Maven Central for Scala 2.10 and 2.11:

libraryDependencies += "io.scalac" %% "reactive-rabbit" % "1.0.1"

Example

Akka Streams - 1.0

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import io.scalac.amqp.Connection


// streaming invoices to Accounting Department
val connection = Connection()
val queue = connection.consume(queue = "invoices")
val exchange = connection.publish(exchange = "accounting_department",
  routingKey = "invoices")

implicit val system = ActorSystem()
implicit val mat = ActorMaterializer()

Source(queue).map(_.message).to(Sink(exchange)).run()