AMQP 1.0

The AMQP Connector adds support for AMQP 1.0 to Reactive Messaging.

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

With this connector, you application can:

  • receive messages from an AMQP Broker or Router.

  • send Message to an AMQP address

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

Using the AMQP connector

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

<dependency>
  <groupId>io.smallrye.reactive</groupId>
  <artifactId>smallrye-reactive-messaging-amqp</artifactId>
  <version>3.3.2</version>
</dependency>

The connector name is: smallrye-amqp.

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

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

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

To use RabbitMQ, refer to Using RabbitMQ.

Receiving messages from AMQP

The AMQP connector lets you retrieve messages from an AMQP broker or router. The AMQP connector retrieves AMQP Messages and maps each of them into Reactive Messaging Messages.

Example

Let’s imagine you have an AMQP broker (such as Apache ActiveMQ Artemis) running, and accessible using the amqp:5672 address (by default it would use localhost:5672). Configure your application to receive AMQP Messages on the prices channel as follows:

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

mp.messaging.incoming.prices.connector=smallrye-amqp  (5)

  1. Configures the broker/router host name. You can do it per channel (using the host attribute) or globally using amqp-host

  2. Configures the broker/router port. You can do it per channel (using the port attribute) or globally using amqp-port. The default is 5672.

  3. Configures the broker/router username if required. You can do it per channel (using the username attribute) or globally using amqp-username.

  4. Configures the broker/router password if required. You can do it per channel (using the password attribute) or globally using amqp-password.

  5. Instructs the prices channel to be managed by the AMQP connector

You don’t need to set the AMQP address. By default, it uses the channel name (prices). You can configure the address attribute to override it.

Then, your application receives Message<Double>. You can consumes the payload directly:

package inbound;

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

import javax.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class AmqpPriceConsumer {

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

}

Or, you can retrieve the Message<Double>:

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 AmqpPriceMessageConsumer {

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

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

}

Deserialization

The connector converts incoming AMQP Messages into Reactive Messaging Message<T> instances. T depends on the body of the received AMQP Message.

The AMQP Type System defines the supported types.

AMQP Body Type <T>

AMQP Value containing a AMQP Primitive Type

the corresponding Java type

AMQP Value using the Binary type

byte[]

AMQP Sequence

List

AMQP Data (with binary content) and the content-type is set to application/json

JsonObject

AMQP Data with a different content-type

byte[]

If you send objects with this AMQP connector (outbound connector), it gets encoded as JSON and sent as binary. The content-type is 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-amqp")
    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-amqp")
    public void consume(JsonObject p) {   (2)
        Price price = p.mapTo(Price.class);  (3)
        prices.add(price);
    }

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

  1. The Price instances are automatically encoded to JSON by the connector

  2. You can receive it using a JsonObject

  3. Then, you can reconstruct the instance using the mapTo method

Inbound Metadata

Messages coming from AMQP contains an instance of IncomingAmqpMetadata in the metadata.

Optional<IncomingAmqpMetadata> metadata = incoming.getMetadata(IncomingAmqpMetadata.class);
metadata.ifPresent(meta -> {
    String address = meta.getAddress();
    String subject = meta.getSubject();
    boolean durable = meta.isDurable();
    // Use io.vertx.core.json.JsonObject
    JsonObject properties = meta.getProperties();
    // ...
});

Acknowledgement

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

Failure Management

If a message produced from an AMQP message is nacked, a failure strategy is applied. The AMQP connector supports six strategies:

  • fail - fail the application; no more AMQP messages will be processed (default). The AMQP message is marked as rejected.

  • accept - this strategy marks the AMQP message as accepted. The processing continues ignoring the failure. Refer to the accepted delivery state documentation.

  • release - this strategy marks the AMQP message as released. The processing continues with the next message. The broker can redeliver the message. Refer to the released delivery state documentation.

  • reject - this strategy marks the AMQP message as rejected. The processing continues with the next message. Refer to the rejected delivery state documentation.

  • modified-failed - this strategy marks the AMQP message as modified and indicates that it failed (with the delivery-failed attribute). The processing continues with the next message, but the broker may attempt to redeliver the message. Refer to the modified delivery state documentation

  • modified-failed-undeliverable-here - this strategy marks the AMQP message as modified and indicates that it failed (with the delivery-failed attribute). It also indicates that the application cannot process the message, meaning that the broker will not attempt to redeliver the message to this node. The processing continues with the next message. Refer to the modified delivery state documentation

Configuration Reference

Table 1. Incoming Attributes of the 'smallrye-amqp' connector
Attribute (alias) Description Mandatory Default

username

(amqp-username)

The username used to authenticate to the broker

Type: string

false

password

(amqp-password)

The password used to authenticate to the broker

Type: string

false

host

(amqp-host)

The broker hostname

Type: string

false

localhost

port

(amqp-port)

The broker port

Type: int

false

5672

use-ssl

(amqp-use-ssl)

Whether the AMQP connection uses SSL/TLS

Type: boolean

false

false

virtual-host

(amqp-virtual-host)

If set, configure the hostname value used for the connection AMQP Open frame and TLS SNI server name (if TLS is in use)

Type: string

false

sni-server-name

(amqp-sni-server-name)

If set, explicitly override the hostname to use for the TLS SNI server name

Type: string

false

reconnect-attempts

(amqp-reconnect-attempts)

The number of reconnection attempts

Type: int

false

100

reconnect-interval

(amqp-reconnect-interval)

The interval in second between two reconnection attempts

Type: int

false

10

connect-timeout

(amqp-connect-timeout)

The connection timeout in milliseconds

Type: int

false

1000

container-id

The AMQP container id

Type: string

false

address

The AMQP address. If not set, the channel name is used

Type: string

false

link-name

The name of the link. If not set, the channel name is used.

Type: string

false

client-options-name

(amqp-client-options-name)

The name of the AMQP Client Option bean used to customize the AMQP client configuration

Type: string

false

broadcast

Whether the received AMQP messages must be dispatched to multiple subscribers

Type: boolean

false

false

durable

Whether AMQP subscription is durable

Type: boolean

false

false

auto-acknowledgement

Whether the received AMQP messages must be acknowledged when received

Type: boolean

false

false

failure-strategy

Specify the failure strategy to apply when a message produced from an AMQP message is nacked. Accepted values are fail (default), accept, release, reject, modified-failed, modified-failed-undeliverable-here

Type: string

false

fail

You can also pass any property supported by the Vert.x AMQP client as attribute.

To use an existing address or queue, you need to configure the address, container-id and, optionally, the link-name attributes. For example, if you have an Apache Artemis broker configured with:

<queues>
    <queue name="people">
        <address>people</address>
        <durable>true</durable>
        <user>artemis</user>
    </queue>
</queues>

You need the following configuration:

mp.messaging.incoming.people.connector=smallrye-amqp
mp.messaging.incoming.people.durable=true
mp.messaging.incoming.people.address=people
mp.messaging.incoming.people.container-id=people

You may need to configure the link-name attribute, if the queue name is not the channel name:

mp.messaging.incoming.people-in.connector=smallrye-amqp
mp.messaging.incoming.people-in.durable=true
mp.messaging.incoming.people-in.address=people
mp.messaging.incoming.people-in.container-id=people
mp.messaging.incoming.people-in.link-name=people

Sending messages to AMQP

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

Example

Let’s imagine you have an AMQP broker (such as Apache ActiveMQ Artemis) running, and accessible using the amqp:5672 address (by default it would use localhost:5672). Configure your application to send the messages from the prices channel as AMQP Message as follows:

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

mp.messaging.outgoing.prices.connector=smallrye-amqp       (5)

  1. Configures the broker/router host name. You can do it per channel (using the host attribute) or globally using amqp-host

  2. Configures the broker/router port. You can do it per channel (using the port attribute) or globally using amqp-port. The default is 5672.

  3. Configures the broker/router username if required. You can do it per channel (using the username attribute) or globally using amqp-username.

  4. Configures the broker/router password if required. You can do it per channel (using the password attribute) or globally using amqp-password.

  5. Instructs the prices channel to be managed by the AMQP connector

You don’t need to set the address. By default, it uses the channel name (prices). You can configure the address attribute to override it.

Then, your application must 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 AmqpPriceProducer {

    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>:

package outbound;

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

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

@ApplicationScoped
public class AmqpPriceMessageProducer {

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

}

Serialization

When receiving a Message<T>, the connector convert the message into an AMQP Message. The payload is converted to the AMQP Message body.

T AMQP Message Body

primitive types or String

AMQP Value with the payload

Instant or UUID

AMQP Value using the corresponding AMQP Type

JsonObject or JsonArray

AMQP Data using a binary content. The content-type is set to application/json

io.vertx.mutiny.core.buffer.Buffer

AMQP Data using a binary content. No content-type set

Any other class

The payload is converted to JSON (using a Json Mapper). The result is wrapped into AMQP Data using a binary content. The content-type is set to application/json

Outbound Metadata

When sending Messages, you can add an instance of OutgoingAmqpMetadata to influence how the message is going to sent to AMQP. For example, you can configure the subjects, properties:

        OutgoingAmqpMetadata metadata = OutgoingAmqpMetadata.builder()
            .withDurable(true)
            .withSubject("my-subject")
            .build();

        // Create a new message from the `incoming` message
        // Add `metadata` to the metadata from the `incoming` message.
        return incoming.addMetadata(metadata);

Dynamic address names

Sometimes it is desirable to select the destination of a message dynamically. In this case, you should not configure the address inside your application configuration file, but instead, use the outbound metadata to set the address.

For example, you can send to a dynamic address based on the incoming message:

        String addressName = selectAddressFromIncommingMessage(incoming);
        OutgoingAmqpMetadata metadata = OutgoingAmqpMetadata.builder()
            .withAddress(addressName)
            .withDurable(true)
            .build();

        // Create a new message from the `incoming` message
        // Add `metadata` to the metadata from the `incoming` message.
        return incoming.addMetadata(metadata);
To be able to set the address per message, the connector is using an anonymous sender.

Acknowledgement

By default, the Reactive Messaging Message is acknowledged when the broker acknowledged the message. When using routers, this acknowledgement may not be enabled. In this case, configure the auto-acknowledgement attribute to acknowledge the message as soon as it has been sent to the router.

If an AMQP message is rejected/released/modified by the broker (or cannot be sent successfully), the message is nacked.

Back Pressure and Credits

The back-pressure is handled by AMQP credits. The outbound connector only requests the amount of allowed credits. When the amount of credits reaches 0, it waits (in a non-blocking fashion) until the broker grants more credits to the AMQP sender.

Configuration Reference

Table 2. Outgoing Attributes of the 'smallrye-amqp' connector
Attribute (alias) Description Mandatory Default

address

The AMQP address. If not set, the channel name is used

Type: string

false

client-options-name

(amqp-client-options-name)

The name of the AMQP Client Option bean used to customize the AMQP client configuration

Type: string

false

connect-timeout

(amqp-connect-timeout)

The connection timeout in milliseconds

Type: int

false

1000

container-id

The AMQP container id

Type: string

false

credit-retrieval-period

The period (in milliseconds) between two attempts to retrieve the credits granted by the broker. This time is used when the sender run out of credits.

Type: int

false

2000

durable

Whether sent AMQP messages are marked durable

Type: boolean

false

false

host

(amqp-host)

The broker hostname

Type: string

false

localhost

link-name

The name of the link. If not set, the channel name is used.

Type: string

false

merge

Whether the connector should allow multiple upstreams

Type: boolean

false

false

password

(amqp-password)

The password used to authenticate to the broker

Type: string

false

port

(amqp-port)

The broker port

Type: int

false

5672

reconnect-attempts

(amqp-reconnect-attempts)

The number of reconnection attempts

Type: int

false

100

reconnect-interval

(amqp-reconnect-interval)

The interval in second between two reconnection attempts

Type: int

false

10

sni-server-name

(amqp-sni-server-name)

If set, explicitly override the hostname to use for the TLS SNI server name

Type: string

false

ttl

The time-to-live of the send AMQP messages. 0 to disable the TTL

Type: long

false

0

use-anonymous-sender

Whether or not the connector should use an anonymous sender.

Type: boolean

false

true

use-ssl

(amqp-use-ssl)

Whether the AMQP connection uses SSL/TLS

Type: boolean

false

false

username

(amqp-username)

The username used to authenticate to the broker

Type: string

false

virtual-host

(amqp-virtual-host)

If set, configure the hostname value used for the connection AMQP Open frame and TLS SNI server name (if TLS is in use)

Type: string

false

You can also pass any property supported by the Vert.x AMQP client as attribute.

To use an existing address or queue, you need to configure the address, container-id and, optionally, the link-name attributes. For example, if you have an Apache Artemis broker configured with:

<queues>
    <queue name="people">
        <address>people</address>
        <durable>true</durable>
        <user>artemis</user>
    </queue>
</queues>

You need the following configuration:

mp.messaging.outgoing.people.connector=smallrye-amqp
mp.messaging.outgoing.people.durable=true
mp.messaging.outgoing.people.address=people
mp.messaging.outgoing.people.container-id=people

You may need to configure the link-name attribute, if the queue name is not the channel name:

mp.messaging.outgoing.people-out.connector=smallrye-amqp
mp.messaging.outgoing.people-out.durable=true
mp.messaging.outgoing.people-out.address=people
mp.messaging.outgoing.people-out.container-id=people
mp.messaging.outgoing.people-out.link-name=people

Customizing the underlying AMQP client

You can customize the underlying AMQP Client configuration by producing an instance of AmqpClientOptions:

@Produces
@Identifier("my-named-options")
public AmqpClientOptions 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 new AmqpClientOptions()
        .setSsl(true)
        .setPemKeyCertOptions(keycert)
        .setPemTrustOptions(trust)
        .addEnabledSaslMechanism("EXTERNAL")
        .setHostnameVerificationAlgorithm("")
        .setConnectTimeout(30000)
        .setReconnectInterval(5000)
        .setContainerId("my-container");
}

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 AMQP connector reports the readiness and liveness of each channel managed by the connector. The AMQP connector uses the same logic for the readiness and liveness checks.

To disable health reporting, set the health-enabled attribute for the channel to false.

On the inbound side (receiving messages from AMQP), the check verifies that the receiver is attached to the broker. On the outbound side (sending records to AMQP), the check verifies that the sender is attached to the broker.

Note that a message processing failures nacks the message, which is then handled by the failure-strategy. It the responsibility of the failure-strategy to report the failure and influence the outcome of the checks. The fail failure strategy reports the failure, and so the check will report the fault.

Using RabbitMQ

This connector is for AMQP 1.0. RabbitMQ implements AMQP 0.9.1. RabbitMQ does not provide AMQP 1.0 by default, but there is a plugin for it. To use RabbitMQ with this connector, enable and configure the AMQP 1.0 plugin.

Despite the plugin, a few features won’t work with RabbitMQ. Thus, we recommend the following configurations.

To receive messages from RabbitMQ:

  • Set durable to false

mp.messaging.incoming.prices.connector=smallrye-amqp
mp.messaging.incoming.prices.durable=false

To send messages to RabbitMQ:

  • set the destination address (anonymous sender are not supported)

  • set use-anonymous-sender to false

mp.messaging.outgoing.generated-price.connector=smallrye-amqp
mp.messaging.outgoing.generated-price.address=prices
mp.messaging.outgoing.generated-price.use-anonymous-sender=false

As a consequence, it’s not possible to change the destination dynamically (using message metadata) when using RabbitMQ.