A Vert.x client allowing applications to interact with a RabbitMQ broker (AMQP 0.9.1)
This service is experimental and the APIs are likely to change before settling down.
Getting Started
Maven
Add the following dependency to your maven project
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-rabbitmq-client</artifactId>
<version>3.4.0.Beta1</version>
</dependency>
Gradle
Add the following dependency to your gradle project
dependencies {
compile 'io.vertx:vertx-rabbitmq-client:3.4.0.Beta1'
}
Create a client
You can create a client instance as follows using a full amqp uri:
import io.vertx.rabbitmq.RabbitMQClient
def config = [:]
// full amqp uri
config.uri = "amqp://xvjvsrrc:VbuL1atClKt7zVNQha0bnnScbNvGiqgb@moose.rmq.cloudamqp.com/xvjvsrrc"
def client = RabbitMQClient.create(vertx, config)
Or you can also specify individual parameters manually:
import io.vertx.rabbitmq.RabbitMQClient
def config = [:]
// Each parameter is optional
// The default parameter with be used if the parameter is not set
config.user = "user1"
config.password = "password1"
config.host = "localhost"
config.port = 5672
config.virtualHost = "vhost1"
config.connectionTimeout = 60
config.requestedHeartbeat = 60
config.handshakeTimeout = 60
config.requestedChannelMax = 5
config.networkRecoveryInterval = 5
config.automaticRecoveryEnabled = true
def client = RabbitMQClient.create(vertx, config)
Declare exchange with additional config
You can pass additional config parameters to RabbitMQ’s exchangeDeclare method
def config = [:]
config["x-dead-letter-exchange"] = "my.deadletter.exchange"
config["alternate-exchange"] = "my.alternate.exchange"
// ...
client.exchangeDeclare("my.exchange", "fanout", true, false, config, { onResult ->
if (onResult.succeeded()) {
println("Exchange successfully declared with config")
} else {
onResult.cause().printStackTrace()
}
})
Operations
The following are some examples of the operations supported by the RabbitMQService API.
Consult the javadoc/documentation for detailed information on all API methods.
Publish
Publish a message to a queue
def message = [
body:"Hello RabbitMQ, from Vert.x !"
]
client.basicPublish("", "my.queue", message, { pubResult ->
if (pubResult.succeeded()) {
println("Message published !")
} else {
pubResult.cause().printStackTrace()
}
})
Consume
Consume messages from a queue
// Create the event bus handler which messages will be sent to
// Create the event bus handler which messages will be sent to
vertx.eventBus().consumer("my.address", { msg ->
def json = msg.body()
println("Got message: ${json.body}")
})
// Setup the link between rabbitmq consumer and event bus address
client.basicConsume("my.queue", "my.address", { consumeResult ->
if (consumeResult.succeeded()) {
println("RabbitMQ consumer created !")
} else {
consumeResult.cause().printStackTrace()
}
})
Get
Will get a message from a queue
client.basicGet("my.queue", true, { getResult ->
if (getResult.succeeded()) {
def msg = getResult.result()
println("Got message: ${msg.body}")
} else {
getResult.cause().printStackTrace()
}
})
Consume messages without auto-ack
// Create the event bus handler which messages will be sent to
vertx.eventBus().consumer("my.address", { msg ->
def json = msg.body()
println("Got message: ${json.body}")
// ack
client.basicAck(json.deliveryTag, false, { asyncResult ->
})
})
// Setup the link between rabbitmq consumer and event bus address
client.basicConsume("my.queue", "my.address", false, { consumeResult ->
if (consumeResult.succeeded()) {
println("RabbitMQ consumer created !")
} else {
consumeResult.cause().printStackTrace()
}
})
Running the tests
You will need to have RabbitMQ installed and running with default ports on localhost for this to work. <a href="mailto:nscavell@redhat.com">Nick Scavelli</a>