Integrate health support to reactive messaging.

Also fix the durability bug when sending messages to a non-durable AMQP address

bom/application/pom.xml

@@ -43,7 +43,7 @@
         <smallrye-context-propagation.version>1.0.13</smallrye-context-propagation.version>
         <smallrye-reactive-streams-operators.version>1.0.13</smallrye-reactive-streams-operators.version>
         <smallrye-converter-api.version>1.1.0</smallrye-converter-api.version>
-        <smallrye-reactive-messaging.version>2.1.1</smallrye-reactive-messaging.version>
+        <smallrye-reactive-messaging.version>2.2.0</smallrye-reactive-messaging.version>
         <swagger-ui.version>3.28.0</swagger-ui.version>
         <jakarta.activation.version>1.2.1</jakarta.activation.version>
         <jakarta.annotation-api.version>1.3.5</jakarta.annotation-api.version>
@@ -3601,6 +3601,11 @@
                 <artifactId>smallrye-reactive-messaging-api</artifactId>
                 <version>${smallrye-reactive-messaging.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.smallrye.reactive</groupId>
+                <artifactId>smallrye-reactive-messaging-health</artifactId>
+                <version>${smallrye-reactive-messaging.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.smallrye.reactive</groupId>
                 <artifactId>smallrye-reactive-messaging-mqtt</artifactId>

docs/src/main/asciidoc/amqp.adoc

@@ -231,7 +231,6 @@ amqp-password=quarkus
 # Configure the AMQP connector to write to the `prices` address
 mp.messaging.outgoing.generated-price.connector=smallrye-amqp
 mp.messaging.outgoing.generated-price.address=prices
-mp.messaging.outgoing.generated-price.durable=true
 
 # Configure the AMQP connector to read from the `prices` queue
 mp.messaging.incoming.prices.connector=smallrye-amqp

extensions/smallrye-reactive-messaging-amqp/deployment/src/test/resources/application.properties

@@ -1,7 +1,6 @@
 amqp-port=5672
 
 mp.messaging.outgoing.source.connector=smallrye-amqp
-mp.messaging.outgoing.source.durable=true
 
 mp.messaging.incoming.in.connector=smallrye-amqp
 mp.messaging.incoming.in.address=source

extensions/smallrye-reactive-messaging/deployment/pom.xml

@@ -27,6 +27,10 @@
       <groupId>io.quarkus</groupId>
       <artifactId>quarkus-smallrye-reactive-messaging</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-smallrye-health-spi</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.quarkus</groupId>
       <artifactId>quarkus-mutiny-reactive-streams-operators-deployment</artifactId>

extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/ReactiveMessagingBuildTimeConfig.java

@@ -0,0 +1,14 @@
+package io.quarkus.smallrye.reactivemessaging.deployment;
+
+import io.quarkus.runtime.annotations.ConfigItem;
+import io.quarkus.runtime.annotations.ConfigPhase;
+import io.quarkus.runtime.annotations.ConfigRoot;
+
+@ConfigRoot(name = "reactive-messaging", phase = ConfigPhase.BUILD_TIME)
+public class ReactiveMessagingBuildTimeConfig {
+    /**
+     * Whether or not an health check is published in case the smallrye-health extension is present.
+     */
+    @ConfigItem(name = "health.enabled", defaultValue = "true")
+    public boolean healthEnabled;
+}

extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/SmallRyeReactiveMessagingProcessor.java

@@ -60,13 +60,16 @@
 import io.quarkus.gizmo.MethodCreator;
 import io.quarkus.gizmo.MethodDescriptor;
 import io.quarkus.gizmo.ResultHandle;
+import io.quarkus.smallrye.health.deployment.spi.HealthBuildItem;
 import io.quarkus.smallrye.reactivemessaging.runtime.QuarkusMediatorConfiguration;
 import io.quarkus.smallrye.reactivemessaging.runtime.QuarkusWorkerPoolRegistry;
 import io.quarkus.smallrye.reactivemessaging.runtime.ReactiveMessagingConfiguration;
 import io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle;
 import io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingRecorder;
 import io.smallrye.reactive.messaging.Invoker;
 import io.smallrye.reactive.messaging.annotations.Blocking;
+import io.smallrye.reactive.messaging.health.SmallRyeReactiveMessagingLivenessCheck;
+import io.smallrye.reactive.messaging.health.SmallRyeReactiveMessagingReadinessCheck;
 
 /**
  *
@@ -302,6 +305,14 @@ public void transform(AnnotationsTransformer.TransformationContext ctx) {
         }
     }
 
+    @BuildStep
+    public void enableHealth(ReactiveMessagingBuildTimeConfig buildTimeConfig, BuildProducer<HealthBuildItem> producer) {
+        producer.produce(
+                new HealthBuildItem(SmallRyeReactiveMessagingLivenessCheck.class.getName(), buildTimeConfig.healthEnabled));
+        producer.produce(
+                new HealthBuildItem(SmallRyeReactiveMessagingReadinessCheck.class.getName(), buildTimeConfig.healthEnabled));
+    }
+
     @BuildStep
     @Record(STATIC_INIT)
     public void build(SmallRyeReactiveMessagingRecorder recorder, RecorderContext recorderContext,

extensions/smallrye-reactive-messaging/runtime/pom.xml

@@ -43,6 +43,32 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>io.smallrye.reactive</groupId>
+            <artifactId>smallrye-reactive-messaging-health</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                    <artifactId>jackson-core</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>javax.annotation</groupId>
+                    <artifactId>javax.annotation-api</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>javax.inject</groupId>
+                    <artifactId>javax.inject</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.osgi</groupId>
+                    <artifactId>org.osgi.annotation.versioning</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>javax.enterprise</groupId>
+                    <artifactId>cdi-api</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>jakarta.annotation</groupId>
             <artifactId>jakarta.annotation-api</artifactId>

integration-tests/reactive-messaging-amqp/src/main/resources/application.properties

@@ -7,7 +7,6 @@ amqp-username=artemis
 amqp-password=artemis
 
 mp.messaging.outgoing.people-out.connector=smallrye-amqp
-mp.messaging.outgoing.people-out.durable=true
 mp.messaging.outgoing.people-out.address=people
 mp.messaging.incoming.people-in.connector=smallrye-amqp
 mp.messaging.incoming.people-in.durable=true