Receiving messages from RabbitMQ

The RabbitMQ connector lets you retrieve messages from a RabbitMQ broker. The RabbitMQ connector retrieves RabbitMQ Messages and maps each of them into Reactive Messaging Messages.

Note

In this context, the reactive messaging concept of a Channel is realised as a RabbitMQ Queue.

Example

Let’s imagine you have a RabbitMQ broker running, and accessible using the rabbitmq:5672 address (by default it would use localhost:5672). Configure your application to receive RabbitMQ Messages on the prices channel as follows:

rabbitmq-host=rabbitmq  # <1>
rabbitmq-port=5672      # <2>
rabbitmq-username=my-username   # <3>
rabbitmq-password=my-password   # <4>

mp.messaging.incoming.prices.connector=smallrye-rabbitmq # <5>
mp.messaging.incoming.prices.queue.name=my-queue         # <6>
mp.messaging.incoming.prices.routing-keys=urgent         # <7>

Configures the broker/router host name. You can do it per channel (using the host attribute) or globally using rabbitmq-host.
Configures the broker/router port. You can do it per channel (using the port attribute) or globally using rabbitmq-port. The default is 5672.
Configures the broker/router username if required. You can do it per channel (using the username attribute) or globally using rabbitmq-username.
Configures the broker/router password if required. You can do it per channel (using the password attribute) or globally using rabbitmq-password.
Instructs the prices channel to be managed by the RabbitMQ connector.
Configures the RabbitMQ queue to read messages from.
Configures the binding between the RabbitMQ exchange and the RabbitMQ queue using a routing key. The default is # (all messages will be forwarded from the exchange to the queue) but in general this can be a comma-separated list of one or more keys.

Then, your application receives Message<String>. You can consume the payload directly:

package rabbitmq.inbound;

import javax.enterprise.context.ApplicationScoped;

import org.eclipse.microprofile.reactive.messaging.Incoming;

@ApplicationScoped
public class RabbitMQPriceConsumer {

    @Incoming("prices")
    public void consume(String price) {
        // process your price.
    }

}

Or, you can retrieve the Message<String>:

package rabbitmq.inbound;

import java.util.concurrent.CompletionStage;

import javax.enterprise.context.ApplicationScoped;

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Message;

@ApplicationScoped
public class RabbitMQPriceMessageConsumer {

    @Incoming("prices")
    public CompletionStage<Void> consume(Message<String> price) {
        // process your price.

        // Acknowledge the incoming message, marking the RabbitMQ message as `accepted`.
        return price.ack();
    }

}

Note

Whether you need to explicitly acknowledge the message depends on the auto-acknowledgement channel setting; if that is set to true then your message will be automatically acknowledged on receipt.

Deserialization

The connector converts incoming RabbitMQ Messages into Reactive Messaging Message<T> instances. The payload type T depends on the value of the RabbitMQ received message Envelope content_type and content_encoding properties.

content_encoding	content_type	Type
Value present	n/a	`byte[]`
No value	`text/plain`	`String`
No value	`application/json`	a JSON element which can be a `JsonArray`, `JsonObject`, `String`, ... if the buffer contains an array, object, string,...
No value	Anything else	`byte[]`

If you send objects with this RabbitMQ connector (outbound connector), they are encoded as JSON and sent with content_type set to application/json. You can receive this payload using (Vert.x) JSON Objects, and then map it to the object class you want:

@ApplicationScoped
public static class Generator {

    @Outgoing("to-rabbitmq")
    public Multi<Price> prices() {             // <1>
        AtomicInteger count = new AtomicInteger();
        return Multi.createFrom().ticks().every(Duration.ofMillis(1000))
                .map(l -> new Price().setPrice(count.incrementAndGet()))
                .onOverflow().drop();
    }

}

@ApplicationScoped
public static class Consumer {

    List<Price> prices = new CopyOnWriteArrayList<>();

    @Incoming("from-rabbitmq")
    public void consume(JsonObject p) {      // <2>
        Price price = p.mapTo(Price.class);  // <3>
        prices.add(price);
    }

    public List<Price> list() {
        return prices;
    }
}

The Price instances are automatically encoded to JSON by the connector
You can receive it using a JsonObject
Then, you can reconstruct the instance using the mapTo method

Inbound Metadata

Messages coming from RabbitMQ contain an instance of IncomingRabbitMQMetadata in the metadata.

RabbitMQ message headers can be accessed from the metadata either by calling getHeader(String header, Class<T> type) to retrieve a single header value. or getHeaders() to get a map of all header values.

final Optional<IncomingRabbitMQMetadata> metadata = incomingMessage.getMetadata(IncomingRabbitMQMetadata.class);
metadata.ifPresent(meta -> {
    final Optional<String> contentEncoding = meta.getContentEncoding();
    final Optional<String> contentType = meta.getContentType();
    final Optional<String> correlationId = meta.getCorrelationId();
    final Optional<ZonedDateTime> timestamp = meta.getTimestamp(ZoneId.systemDefault());
    final Optional<Integer> priority = meta.getPriority();
    final Optional<String> replyTo = meta.getReplyTo();
    final Optional<String> userId = meta.getUserId();

    // Access a single String-valued header
    final Optional<String> stringHeader = meta.getHeader("my-header", String.class);

    // Access all headers
    final Map<String, Object> headers = meta.getHeaders();
    // ...
});

The type <T> of the header value depends on the RabbitMQ type used for the header:

RabbitMQ Header Type	T
String	`String`
Boolean	`Boolean`
Number	`Number`
List	`java.util.List`

Acknowledgement

When a Reactive Messaging Message associated with a RabbitMQ Message is acknowledged, it informs the broker that the message has been accepted.

Whether you need to explicitly acknowledge the message depends on the auto-acknowledgement setting for the channel; if that is set to true then your message will be automatically acknowledged on receipt.

Failure Management

If a message produced from a RabbitMQ message is nacked, a failure strategy is applied. The RabbitMQ connector supports three strategies, controlled by the failure-strategy channel setting:

fail - fail the application; no more RabbitMQ messages will be processed. The RabbitMQ message is marked as rejected.
accept - this strategy marks the RabbitMQ message as accepted. The processing continues ignoring the failure.
reject - this strategy marks the RabbitMQ message as rejected (default). The processing continues with the next message.

Configuration Reference

Attribute (alias)	Description	Type	Mandatory	Default
auto-acknowledgement	Whether the received RabbitMQ messages must be acknowledged when received; if true then delivery constitutes acknowledgement	boolean	false	`false`
auto-bind-dlq	Whether to automatically declare the DLQ and bind it to the binder DLX	boolean	false	`false`
automatic-recovery-enabled	Whether automatic connection recovery is enabled	boolean	false	`false`
automatic-recovery-on-initial-connection	Whether automatic recovery on initial connections is enabled	boolean	false	`true`
broadcast	Whether the received RabbitMQ messages must be dispatched to multiple subscribers	boolean	false	`false`
client-options-name (rabbitmq-client-options-name)	The name of the RabbitMQ Client Option bean used to customize the RabbitMQ client configuration	string	false
connection-count	The number of RabbitMQ connections to create for consuming from this queue. This might be necessary to consume from a sharded queue with a single client.	int	false	`1`
connection-timeout	The TCP connection timeout (ms); 0 is interpreted as no timeout	int	false	`60000`
content-type-override	Override the content_type attribute of the incoming message, should be a valid MINE type	string	false
credentials-provider-name (rabbitmq-credentials-provider-name)	The name of the RabbitMQ Credentials Provider bean used to provide dynamic credentials to the RabbitMQ client	string	false
dead-letter-exchange	A DLX to assign to the queue. Relevant only if auto-bind-dlq is true	string	false	`DLX`
dead-letter-exchange-type	The type of the DLX to assign to the queue. Relevant only if auto-bind-dlq is true	string	false	`direct`
dead-letter-queue-name	The name of the DLQ; if not supplied will default to the queue name with '.dlq' appended	string	false
dead-letter-routing-key	A dead letter routing key to assign to the queue; if not supplied will default to the queue name	string	false
dlx.declare	Whether to declare the dead letter exchange binding. Relevant only if auto-bind-dlq is true; set to false if these are expected to be set up independently	boolean	false	`false`
exchange.auto-delete	Whether the exchange should be deleted after use	boolean	false	`false`
exchange.declare	Whether to declare the exchange; set to false if the exchange is expected to be set up independently	boolean	false	`true`
exchange.durable	Whether the exchange is durable	boolean	false	`true`
exchange.name	The exchange that messages are published to or consumed from. If not set, the channel name is used	string	false
exchange.type	The exchange type: direct, fanout, headers or topic (default)	string	false	`topic`
failure-strategy	The failure strategy to apply when a RabbitMQ message is nacked. Accepted values are `fail`, `accept`, `reject` (default)	string	false	`reject`
handshake-timeout	The AMQP 0-9-1 protocol handshake timeout (ms)	int	false	`10000`
host (rabbitmq-host)	The broker hostname	string	false	`localhost`
include-properties	Whether to include properties when a broker message is passed on the event bus	boolean	false	`false`
keep-most-recent	Whether to discard old messages instead of recent ones	boolean	false	`false`
max-incoming-internal-queue-size	The maximum size of the incoming internal queue	int	false	`500000`
network-recovery-interval	How long (ms) will automatic recovery wait before attempting to reconnect	int	false	`5000`
password (rabbitmq-password)	The password used to authenticate to the broker	string	false
port (rabbitmq-port)	The broker port	int	false	`5672`
queue.auto-delete	Whether the queue should be deleted after use	boolean	false	`false`
queue.declare	Whether to declare the queue and binding; set to false if these are expected to be set up independently	boolean	false	`true`
queue.durable	Whether the queue is durable	boolean	false	`true`
queue.exclusive	Whether the queue is for exclusive use	boolean	false	`false`
queue.name	The queue from which messages are consumed.	string	true
queue.single-active-consumer	If set to true, only one consumer can actively consume messages	boolean	false
queue.ttl	If specified, the time (ms) for which a message can remain in the queue undelivered before it is dead	long	false
reconnect-attempts (rabbitmq-reconnect-attempts)	The number of reconnection attempts	int	false	`100`
reconnect-interval (rabbitmq-reconnect-interval)	The interval (in seconds) between two reconnection attempts	int	false	`10`
requested-channel-max	The initially requested maximum channel number	int	false	`2047`
requested-heartbeat	The initially requested heartbeat interval (seconds), zero for none	int	false	`60`
routing-keys	A comma-separated list of routing keys to bind the queue to the exchange	string	false	`#`
ssl (rabbitmq-ssl)	Whether or not the connection should use SSL	boolean	false	`false`
tracing.attribute-headers	A comma-separated list of headers that should be recorded as span attributes. Relevant only if tracing.enabled=true	string	false	``
tracing.enabled	Whether tracing is enabled (default) or disabled	boolean	false	`true`
trust-all (rabbitmq-trust-all)	Whether to skip trust certificate verification	boolean	false	`false`
trust-store-password (rabbitmq-trust-store-password)	The password of the JKS trust store	string	false
trust-store-path (rabbitmq-trust-store-path)	The path to a JKS trust store	string	false
use-nio	Whether usage of NIO Sockets is enabled	boolean	false	`false`
user	The user name to use when connecting to the broker	string	false	`guest`
username (rabbitmq-username)	The username used to authenticate to the broker	string	false
virtual-host (rabbitmq-virtual-host)	The virtual host to use when connecting to the broker	string	false	`/`

To use an existing queue, you need to configure the queue.name attribute.

For example, if you have a RabbitMQ broker already configured with a queue called peopleQueue that you wish to read messages from, you need the following configuration:

1 2	`mp.messaging.incoming.people.connector=smallrye-rabbitmq mp.messaging.incoming.people.queue.name=peopleQueue`

If you want RabbitMQ to create the queue for you but bind it to an existing topic exchange people, you need the following configuration:

1
2
3

mp.messaging.incoming.people.connector=smallrye-rabbitmq
mp.messaging.incoming.people.queue.name=peopleQueue
mp.messaging.incoming.people.queue.declare=true

Note

In the above the channel name people is implicitly assumed to be the name of the exchange; if this is not the case you would need to name the exchange explicitly using the exchange.name property.

If you want RabbitMQ to create the people exchange, queue and binding, you need the following configuration:

mp.messaging.incoming.people.connector=smallrye-rabbitmq
mp.messaging.incoming.people.exchange.declare=true
mp.messaging.incoming.people.queue.name=peopleQueue
mp.messaging.incoming.people.queue.declare=true
mp.messaging.incoming.people.queue.routing-keys=tall,short

In the above we have used an explicit list of routing keys rather than the default (#). Each component of the list creates a separate binding between the queue and the exchange, so in the case above we would have two bindings; one based on a routing key of tall, the other based on one of short.

Note

The default value of routing-keys is # (indicating a match against all possible routing keys) which is only appropriate for topic Exchanges. If you are using other types of exchange and/or need to declare queue bindings, you’ll need to supply a valid value for the exchange in question.