Skip to main content
Version: 2.x

ZIO AMQP

ZIO AMQP is a ZIO-based wrapper around the RabbitMQ client. It provides a streaming interface to AMQP queues and helps to prevent you from shooting yourself in the foot with thread-safety issues.

Development CI Badge Sonatype Releases Sonatype Snapshots javadoc ZIO AMQP

Installation

Add the following lines to your build.sbt file:

libraryDependencies += "dev.zio" %% "zio-amqp" % "1.0.0-alpha.3"

The latest version is built against ZIO 2.0.0-RC6.

Consuming

The example below creates a connection to an AMQP server and then creates a channel. Both are created as Managed resources, which means they are closed automatically after using even in the face of errors.

The example then creates a stream of the messages consumed from a queue named "queueName". Each received message is acknowledged back to the AMQP server.

import nl.vroste.zio.amqp._
import nl.vroste.zio.amqp.model._
import java.net.URI
import zio._
import zio.Console._

val channel: ZIO[Scope, Throwable, Channel] = for {
connection <- Amqp.connect(URI.create("amqp://my_amqp_server_uri"))
channel <- Amqp.createChannel(connection)
} yield channel

val effect: ZIO[Any, Throwable, Unit] =
ZIO.scoped {
channel.flatMap { channel =>
channel
.consume(queue = QueueName("queueName"), consumerTag = ConsumerTag("test"))
.mapZIO { record =>
val deliveryTag = record.getEnvelope.getDeliveryTag
printLine(s"Received ${deliveryTag}: ${new String(record.getBody)}") *>
channel.ack(DeliveryTag(deliveryTag))
}
.take(5)
.runDrain
}
}

See the ZIO documentation for more information on how to run this effect or integrate with an existing application.