RabbitMQ

The RabbitMQ Connector adds support for RabbitMQ to Reactive Messaging, based on the AMQP 0-9-1 Protocol Specification.

Advanced Message Queuing Protocol 0-9-1 (AMQP 0-9-1) is an open standard for passing business messages between applications or organizations.

With this connector, your application can:

receive messages from a RabbitMQ queue
send messages to a RabbitMQ exchange

The RabbitMQ connector is based on the Vert.x RabbitMQ Client.

The AMQP connector supports the AMQP 1.0 protocol, which is very different from AMQP 0-9-1. You can use the AMQP connector with RabbitMQ provided that the latter has the AMQP 1.0 Plugin installed, albeit with reduced functionality.

Using the RabbitMQ connector

To use the RabbitMQ Connector, add the following dependency to your project:

<dependency>
  <groupId>io.smallrye.reactive</groupId>
  <artifactId>smallrye-reactive-messaging-rabbitmq</artifactId>
  <version>3.13.0</version>
</dependency>

The connector name is: smallrye-rabbitmq.

So, to indicate that a channel is managed by this connector you need:

# Inbound
mp.messaging.incoming.[channel-name].connector=smallrye-rabbitmq

# Outbound
mp.messaging.outgoing.[channel-name].connector=smallrye-rabbitmq

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.

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 inbound;

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

import javax.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class RabbitMQPriceConsumer {

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

}

Or, you can retrieve the Message<String>:

package inbound;

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

import javax.enterprise.context.ApplicationScoped;
import java.util.concurrent.CompletionStage;

@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();
    }

}

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 T

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

Value present

n/a

byte[]

No value

text/plain

String

No value

application/json

a JSON element which can be a JsonArray, JsonObject, String, …etc if the buffer contains an array, object, string, …etc

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> creationTime = meta.getCreationTime(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

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

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

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

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:

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

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.

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.

Sending messages to RabbitMQ

The RabbitMQ connector can write Reactive Messaging Messages as RabbitMQ Messages.

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

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 send the messages from the prices channel as a RabbitMQ Message as follows:

rabbitmq-host=rabbitmq         (1)
rabbitmq-port=5672             (2)
rabbitmq-username=my-username  (3)
rabbitmq-password=my-password  (4)

mp.messaging.outgoing.prices.connector=smallrye-rabbitmq  (5)
mp.messaging.outgoing.prices.default-routing-key=normal   (6)

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
Supplies the default routing key to be included in outbound messages; this will be if the "raw payload" form of message sending is used (see below).

You don’t need to set the RabbitMQ exchange name. By default, it uses the channel name (prices) as the name of the exchange to send messages to. You can configure the exchange.name attribute to override it.

Then, your application can send Message<Double> to the prices channel. It can use double payloads as in the following snippet:

package outbound;

import io.smallrye.mutiny.Multi;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import javax.enterprise.context.ApplicationScoped;
import java.time.Duration;
import java.util.Random;

@ApplicationScoped
public class RabbitMQPriceProducer {

    private Random random = new Random();

    @Outgoing("prices")
    public Multi<Double> generate() {
        // Build an infinite stream of random prices
        // It emits a price every second
        return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
                .map(x -> random.nextDouble());
    }

}

Or, you can send Message<Double>, which affords the opportunity to explicitly specify metadata on the outgoing message:

package outbound;

import io.smallrye.mutiny.Multi;
import io.smallrye.reactive.messaging.rabbitmq.OutgoingRabbitMQMetadata;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.eclipse.microprofile.reactive.messaging.Metadata;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import javax.enterprise.context.ApplicationScoped;
import java.time.Duration;
import java.time.ZonedDateTime;
import java.util.Random;

@ApplicationScoped
public class RabbitMQPriceMessageProducer {

    private Random random = new Random();

    @Outgoing("prices")
    public Multi<Message<Double>> generate() {
        // Build an infinite stream of random prices
        // It emits a price every second
        return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
                .map(x -> Message.of(random.nextDouble(),
                        Metadata.of(new OutgoingRabbitMQMetadata.Builder()
                                .withRoutingKey("normal")
                                .withTimestamp(ZonedDateTime.now())
                                .build())
                ));
    }

}

Serialization

When sending a Message<T>, the connector converts the message into a RabbitMQ Message. The payload is converted to the RabbitMQ Message body.

T RabbitMQ Message Body

T	RabbitMQ Message Body
primitive types or `UUID`/`String`	String value with `content_type` set to `text/plain`
`JsonObject` or `JsonArray`	Serialized String payload with `content_type` set to `application/json`
`io.vertx.mutiny.core.buffer.Buffer`	Binary content, with `content_type` set to `application/octet-stream`
`byte[]`	Binary content, with content_type set to `application/octet-stream`
Any other class	The payload is converted to JSON (using a Json Mapper) then serialized with `content_type` set to `application/json`

primitive types or UUID/String

String value with content_type set to text/plain

JsonObject or JsonArray

Serialized String payload with content_type set to application/json

io.vertx.mutiny.core.buffer.Buffer

Binary content, with content_type set to application/octet-stream

byte[]

Binary content, with content_type set to application/octet-stream

Any other class

The payload is converted to JSON (using a Json Mapper) then serialized with content_type set to application/json

If the message payload cannot be serialized to JSON, the message is nacked.

Outbound Metadata

When sending Messages, you can add an instance of OutgoingRabbitMQMetadata to influence how the message is handled by RabbitMQ. For example, you can configure the routing key, timestamp and headers:

        final OutgoingRabbitMQMetadata metadata = new OutgoingRabbitMQMetadata.Builder()
                .withHeader("my-header", "xyzzy")
                .withRoutingKey("urgent")
                .withTimestamp(ZonedDateTime.now())
                .build();

        // Add `metadata` to the metadata of the outgoing message.
        return Message.of("Hello", Metadata.of(metadata));

Acknowledgement

By default, the Reactive Messaging Message is acknowledged when the broker acknowledges the message.

Configuration Reference

Table 2. Outgoing Attributes of the 'smallrye-rabbitmq' connector
Attribute (alias)	Description	Mandatory	Default
automatic-recovery-enabled	Whether automatic connection recovery is enabled Type: boolean	false	`false`
automatic-recovery-on-initial-connection	Whether automatic recovery on initial connections is enabled Type: boolean	false	`true`
client-options-name (rabbitmq-client-options-name)	The name of the RabbitMQ Client Option bean used to customize the RabbitMQ client configuration Type: string	false
connection-timeout	The TCP connection timeout (ms); 0 is interpreted as no timeout Type: int	false	`60000`
credentials-provider-name (rabbitmq-credentials-provider-name)	The name of the RabbitMQ Credentials Provider bean used to provide dynamic credentials to the RabbitMQ client Type: string	false
default-routing-key	The default routing key to use when sending messages to the exchange Type: string	false	``
default-ttl	If specified, the time (ms) sent messages can remain in queues undelivered before they are dead Type: long	false
exchange.auto-delete	Whether the exchange should be deleted after use Type: boolean	false	`false`
exchange.declare	Whether to declare the exchange; set to false if the exchange is expected to be set up independently Type: boolean	false	`true`
exchange.durable	Whether the exchange is durable Type: boolean	false	`true`
exchange.name	The exchange that messages are published to or consumed from. If not set, the channel name is used Type: string	false
exchange.type	The exchange type: direct, fanout, headers or topic (default) Type: string	false	`topic`
handshake-timeout	The AMQP 0-9-1 protocol handshake timeout (ms) Type: int	false	`10000`
host (rabbitmq-host)	The broker hostname Type: string	false	`localhost`
include-properties	Whether to include properties when a broker message is passed on the event bus Type: boolean	false	`false`
max-inflight-messages	The maximum number of messages to be written to RabbitMQ concurrently; must be a positive number Type: long	false	`1024`
max-outgoing-internal-queue-size	The maximum size of the outgoing internal queue Type: int	false
network-recovery-interval	How long (ms) will automatic recovery wait before attempting to reconnect Type: int	false	`5000`
password (rabbitmq-password)	The password used to authenticate to the broker Type: string	false
port (rabbitmq-port)	The broker port Type: int	false	`5672`
reconnect-attempts (rabbitmq-reconnect-attempts)	The number of reconnection attempts Type: int	false	`100`
reconnect-interval (rabbitmq-reconnect-interval)	The interval (in seconds) between two reconnection attempts Type: int	false	`10`
requested-channel-max	The initially requested maximum channel number Type: int	false	`2047`
requested-heartbeat	The initially requested heartbeat interval (seconds), zero for none Type: int	false	`60`
ssl (rabbitmq-ssl)	Whether or not the connection should use SSL Type: 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 Type: string	false	``
tracing.enabled	Whether tracing is enabled (default) or disabled Type: boolean	false	`true`
trust-all (rabbitmq-trust-all)	Whether to skip trust certificate verification Type: boolean	false	`false`
trust-store-password (rabbitmq-trust-store-password)	The password of the JKS trust store Type: string	false
trust-store-path (rabbitmq-trust-store-path)	The path to a JKS trust store Type: string	false
use-nio	Whether usage of NIO Sockets is enabled Type: boolean	false	`false`
user	The user name to use when connecting to the broker Type: string	false	`guest`
username (rabbitmq-username)	The username used to authenticate to the broker Type: string	false
virtual-host (rabbitmq-virtual-host)	The virtual host to use when connecting to the broker Type: string	false	`/`

Using existing destinations

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

For example, if you have a RabbitMQ broker already configured with an exchange called people that you wish to send messages to, you need the following configuration:

mp.messaging.outgoing.people.connector=smallrye-rabbitmq

You would need to configure the exchange.name attribute, if the exchange name were not the channel name:

mp.messaging.outgoing.people-out.connector=smallrye-rabbitmq
mp.messaging.outgoing.people-out.exchange.name=people

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

mp.messaging.outgoing.people-out.connector=smallrye-amqp
mp.messaging.outgoing.people-out.exchange.name=people
mp.messaging.outgoing.people-out.exchange.declare=true

The above example will create a topic exchange and use an empty default routing-key (unless overridden programatically using outgoing metadata for the message). If you want to create a different type of exchange or have a different default routing key, then the exchange.type and default-routing-key properties need to be explicitly specified.

Customizing the underlying RabbitMQ client

You can customize the underlying RabbitMQ Client configuration by producing an instance of RabbitMQOptions:

@Produces
@Identifier("my-named-options")
public RabbitMQOptions getNamedOptions() {
    // You can use the produced options to configure the TLS connection
    PemKeyCertOptions keycert = new PemKeyCertOptions()
        .addCertPath("./tls/tls.crt")
        .addKeyPath("./tls/tls.key");
    PemTrustOptions trust =
        new PemTrustOptions().addCertPath("./tlc/ca.crt");

    return (RabbitMQOptions) new RabbitMQOptions()
        .setUser("admin")
        .setPassword("test")
        .setSsl(true)
        .setPemKeyCertOptions(keycert)
        .setPemTrustOptions(trust)
        .setHostnameVerificationAlgorithm("")
        .setConnectTimeout(30000)
        .setReconnectInterval(5000);
}

This instance is retrieved and used to configure the client used by the connector. You need to indicate the name of the client using the client-options-name attribute:

mp.messaging.incoming.prices.client-options-name=my-named-options

Health reporting

The RabbitMQ connector reports the readiness and liveness of each channel managed by the connector.

On the inbound side (receiving messages from RabbitMQ), the check verifies that the receiver is connected to the broker.

On the outbound side (sending records to RabbitMQ), the check verifies that the sender is not disconnected from the broker; the sender may still be in an initiliased state (connection not yet attempted), but this is regarded as live/ready.

Connecting to managed instances

This section describes the connector configuration to use managed RabbitMQ instances (hosted on the Cloud).

Cloud AMQP

To connect to an instance of RabbitMQ hosted on Cloud AMQP, use the following configuration:

rabbitmq-host=host-name
rabbitmq-port=5671
rabbitmq-username=user-name
rabbitmq-password=password
rabbitmq-virtual-host=user-name
rabbitmq-ssl=true

You can extract the values from the AMQPS url displayed on the administration portal:

amqps://user-name:password@host/user-name

Amazon MQ

Amazon MQ can host RabbitMQ brokers (as well as AMQP 1.0 brokers). To connect to a RabbitMQ instance hosted on Amazon MQ, use the following configuration:

rabbitmq-host=host-name
rabbitmq-port=5671
rabbitmq-username=user-name
rabbitmq-password=password
rabbitmq-ssl=true

You can extract the host value from the AMQPS url displayed on the administration console:

amqps://foobarbaz.mq.us-east-2.amazonaws.com:5671

The username and password are configured during the broker creation.