New RabbitMQ connector based on the RabbitMQ Java client

Author: cescoffier

documentation/mkdocs.yml

@@ -67,6 +67,15 @@ nav:
         - 'Client Customization' : rabbitmq/rabbitmq-client-customization.md
         - 'Connecting to managed instances' : rabbitmq/rabbitmq-cloud.md
 
+  - RabbitMQ OG:
+        - rabbitmq-og/rabbitmq-og.md
+        - 'Receiving messages' : rabbitmq-og/receiving-messages-from-rabbitmq.md
+        - 'Sending messages' : rabbitmq-og/sending-messages-to-rabbitmq.md
+        - 'Health Checks' : rabbitmq-og/rabbitmq-og-health.md
+        - 'Client Customization' : rabbitmq-og/rabbitmq-og-client-customization.md
+        - 'OpenTelemetry Tracing' : rabbitmq-og/rabbitmq-og-tracing.md
+        - 'Connecting to managed instances' : rabbitmq-og/rabbitmq-og-cloud.md
+
   - Pulsar:
         - pulsar/pulsar.md
         - 'Receiving messages': pulsar/receiving-pulsar-messages.md

documentation/pom.xml

@@ -72,6 +72,11 @@
       <artifactId>smallrye-reactive-messaging-rabbitmq</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>io.smallrye.reactive</groupId>
+      <artifactId>smallrye-reactive-messaging-rabbitmq-og</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>io.smallrye.reactive</groupId>
       <artifactId>smallrye-reactive-messaging-pulsar</artifactId>

documentation/src/main/docs/rabbitmq-og/rabbitmq-og-client-customization.md

@@ -0,0 +1,49 @@
+# Customizing the underlying RabbitMQ client
+
+You can customize the underlying RabbitMQ Client configuration by
+*producing* an instance of
+[`ConnectionFactory`](https://rabbitmq.github.io/rabbitmq-java-client/api/current/com/rabbitmq/client/ConnectionFactory.html):
+
+``` java
+{{ insert('rabbitmq/og/customization/RabbitMQProducers.java', 'named') }}
+```
+
+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
+
+## Credentials Provider
+
+The OG connector supports RabbitMQ's
+[`CredentialsProvider`](https://rabbitmq.github.io/rabbitmq-java-client/api/current/com/rabbitmq/client/impl/CredentialsProvider.html)
+interface for dynamic credential management. This is useful when
+credentials are rotated or fetched from an external secrets manager.
+
+To use a credentials provider, expose a CDI bean implementing
+`com.rabbitmq.client.impl.CredentialsProvider` with an `@Identifier`
+qualifier, and reference it via the `credentials-provider-name` attribute:
+
+```properties
+mp.messaging.incoming.prices.credentials-provider-name=my-credentials-provider
+```
+
+## Cluster mode
+
+To connect to a RabbitMQ cluster, use the `addresses` attribute to
+specify multiple broker addresses. When set, this overrides the `host`
+and `port` attributes:
+
+```properties
+rabbitmq-addresses=host1:5672,host2:5672,host3:5672
+```
+
+## NIO Sockets
+
+The connector supports using NIO sockets for the RabbitMQ connection.
+Enable NIO mode with:
+
+```properties
+rabbitmq-use-nio=true
+```

documentation/src/main/docs/rabbitmq-og/rabbitmq-og-cloud.md

@@ -0,0 +1,44 @@
+# 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](https://www.cloudamqp.com/), use the following configuration:
+
+``` properties
+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](https://aws.amazon.com/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:
+
+``` properties
+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.

documentation/src/main/docs/rabbitmq-og/rabbitmq-og-health.md

@@ -0,0 +1,25 @@
+# Health reporting
+
+The RabbitMQ OG 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 initialized state (connection not yet attempted), but
+this is regarded as live/ready.
+
+You can disable health reporting by setting the `health-enabled` attribute of the channel to `false`.
+It disables both liveness and readiness.
+You can disable readiness reporting by setting the `health-readiness-enabled` attribute of the channel to `false`.
+
+## @Channel and lazy subscription
+
+When you inject a channel using `@Channel` annotation, you are responsible for subscribing to the channel.
+Until the subscription happens, the channel is not connected to the broker and thus cannot receive messages.
+The default health check will fail in this case.
+
+To handle this use case, you need to configure the `health-lazy-subscription` attribute of the channel to `true`.
+It configures the health check to not fail if there are no subscription yet.

documentation/src/main/docs/rabbitmq-og/rabbitmq-og-tracing.md

@@ -0,0 +1,38 @@
+# OpenTelemetry Tracing
+
+The RabbitMQ OG connector supports [OpenTelemetry](https://opentelemetry.io/)
+tracing for both incoming and outgoing channels.
+
+## Enabling tracing
+
+Tracing is enabled by default. You can disable it per channel with:
+
+```properties
+mp.messaging.incoming.prices.tracing.enabled=false
+mp.messaging.outgoing.prices.tracing.enabled=false
+```
+
+## How it works
+
+When tracing is enabled:
+
+-   **Outgoing messages**: The connector creates a `PUBLISH` span and
+    injects the trace context into the message headers before sending.
+
+-   **Incoming messages**: The connector extracts the trace context from
+    the message headers and creates a `RECEIVE` span linked to the
+    producer's trace context.
+
+The span attributes include the exchange name and routing key.
+
+## Including headers as span attributes
+
+You can configure specific message headers to be recorded as span
+attributes using the `tracing.attribute-headers` property:
+
+```properties
+mp.messaging.incoming.prices.tracing.attribute-headers=my-header,another-header
+```
+
+This is a comma-separated list of header names whose values will be
+added as attributes to the tracing span.

documentation/src/main/docs/rabbitmq-og/rabbitmq-og.md

@@ -0,0 +1,55 @@
+# RabbitMQ OG Connector
+
+The RabbitMQ OG 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](https://www.rabbitmq.com/resources/specs/amqp0-9-1.pdf)) 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 OG connector is based on the [RabbitMQ Java Client](https://www.rabbitmq.com/client-libraries/java-client),
+the official RabbitMQ client library.
+
+!!!note
+    This connector is an alternative to the existing RabbitMQ connector
+    (`smallrye-rabbitmq`), which is based on the Vert.x RabbitMQ client.
+    The OG connector uses the original RabbitMQ Java client directly,
+    providing improved reconnection handling and direct access to all
+    RabbitMQ client features.
+
+!!!important
+    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](https://github.com/rabbitmq/rabbitmq-amqp1.0/blob/v3.7.x/README.md)
+    installed, albeit with reduced functionality.
+
+## Using the RabbitMQ OG connector
+
+To use the RabbitMQ OG Connector, add the following dependency to your
+project:
+
+``` xml
+<dependency>
+  <groupId>io.smallrye.reactive</groupId>
+  <artifactId>smallrye-reactive-messaging-rabbitmq-og</artifactId>
+  <version>{{ attributes['project-version'] }}</version>
+</dependency>
+```
+
+The connector name is: `smallrye-rabbitmq-og`.
+
+So, to indicate that a channel is managed by this connector you need:
+```properties
+# Inbound
+mp.messaging.incoming.[channel-name].connector=smallrye-rabbitmq-og
+
+# Outbound
+mp.messaging.outgoing.[channel-name].connector=smallrye-rabbitmq-og
+```