Health support

spec/src/main/asciidoc/architecture.asciidoc

@@ -1,5 +1,5 @@
 //
-// Copyright (c) 2018, 2020 Contributors to the Eclipse Foundation
+// Copyright (c) 2018, 2021 Contributors to the Eclipse Foundation
 //
 // See the NOTICE file(s) distributed with this work for additional
 // information regarding copyright ownership.
@@ -1331,6 +1331,44 @@ The connector is responsible for the acknowledgment (positive or negative) of th
 * An outgoing connector must acknowledge the incoming `org.eclipse.microprofile.reactive.messaging.Message` once it has successfully dispatched the message.
 * An outgoing connector must acknowledge negatively the incoming `org.eclipse.microprofile.reactive.messaging.Message` if it cannot be dispatched.
 
+== Health
+
+When MicroProfile Reactive Messaging is used in an environment where MicroProfile Health is enabled, implicit readiness
+and liveness checks `messaging` are produced.
+
+[source, json]
+----
+{
+  "status": "UP",
+  "checks": [
+    {
+      "name": "messaging",
+      "status": "UP"
+    }
+  ]
+}
+----
+
+=== Readiness
+State of the readiness check `messaging` is DOWN until subscribers of all messaging channels are subscribed to their
+respective publishers and `onSubscribe` signal is detected on all channels.
+
+=== Liveness
+
+State of the liveness check `messaging` is UP until `cancel` or `onError` signals are detected on any of the messaging
+channels.
+
+=== Excluding channels
+
+For channels which are not desired to be checked for liveness or readiness, exclusion is configurable with
+`mp.messaging.health.live.exclude` or `mp.messaging.health.ready.exclude` config keys.
+
+[source, properties]
+----
+mp.messaging.health.live.exclude=<channel-name>,<channel-name>
+mp.messaging.health.ready.exclude=<channel-name>,<channel-name>
+----
+
 == Metrics
 
 When MicroProfile Reactive Messaging is used in an environment where MicroProfile Metrics is enabled, the Reactive Messaging implementation automatically produces metrics.

tck/pom.xml

@@ -34,6 +34,7 @@
     <properties>
         <arquillian.version>1.4.1.Final</arquillian.version>
         <shrinkwrap-api.version>1.2.6</shrinkwrap-api.version>
+        <version.glassfish.json>1.1.6</version.glassfish.json>
     </properties>
 
     <dependencyManagement>
@@ -120,6 +121,19 @@
             <artifactId>awaitility</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>jakarta.json</groupId>
+            <artifactId>jakarta.json-api</artifactId>
+            <version>${version.glassfish.json}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.glassfish</groupId>
+            <artifactId>jakarta.json</artifactId>
+            <version>${version.glassfish.json}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.jboss.shrinkwrap</groupId>
             <artifactId>shrinkwrap-api</artifactId>

tck/src/main/java/org/eclipse/microprofile/reactive/messaging/tck/health/ChannelRegister.java

@@ -0,0 +1,184 @@
+/**
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * You may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.eclipse.microprofile.reactive.messaging.tck.health;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+import javax.enterprise.context.ApplicationScoped;
+
+import org.junit.Assert;
+
+/**
+ * Helper bean for keeping references of all the messaging channels
+ * used in the health test scenario.
+ */
+@ApplicationScoped
+public class ChannelRegister {
+
+    private final Map<String, Channel<?>> channelMap = new HashMap<>();
+    private final ReentrantLock mapLock = new ReentrantLock();
+
+    /**
+     * Return representation of messaging channel by its name.
+     *
+     * @param channelName name of the messaging channel
+     * @param <P>         payload type of the channel
+     * @return the channel with references to its publisher and subscriber
+     */
+    @SuppressWarnings("unchecked")
+    <P> Channel<P> get(String channelName) {
+        try {
+            mapLock.lock();
+            channelMap.putIfAbsent(channelName, new Channel<P>());
+            return (Channel<P>) channelMap.get(channelName);
+        }
+        finally {
+            mapLock.unlock();
+        }
+    }
+
+    /**
+     * Return all the channels used in the health test scenario.
+     *
+     * @return all the channels
+     */
+    Collection<Channel<?>> getAllChannels() {
+        try {
+            mapLock.lock();
+            return Collections.unmodifiableCollection(channelMap.values());
+        }
+        finally {
+            mapLock.unlock();
+        }
+    }
+
+    /**
+     * Iterate over all registered channels, on each's publisher invoke onSubscribe signal
+     * and block until any onSubscribe signal is received on its subscriber.
+     */
+    void readyAll() {
+        getAllChannels().forEach(Channel::ready);
+    }
+
+    /**
+     * Iterate over all registered channels, on each's publisher invoke onError signal
+     * and block until any error signal is received on its subscriber.
+     */
+    void failAll() {
+        getAllChannels().forEach(Channel::fail);
+    }
+
+    /**
+     * Iterate over all registered channels, on each's publisher invoke onComplete signal
+     * and block until any complete signal is received on its subscriber.
+     */
+    void completeAll() {
+        getAllChannels().forEach(Channel::complete);
+    }
+
+    /**
+     * Iterate over all registered channels, on each's subscriber subscription invoke cancel signal
+     * and block until any cancel signal is received on its publisher.
+     */
+    void cancelAll() {
+        getAllChannels().forEach(Channel::cancel);
+    }
+
+    /**
+     * Represents a messaging channel used in the health test scenario.
+     * @param <PAYLOAD> type of the channel's payload
+     */
+    public static class Channel<PAYLOAD> {
+        private final TestPublisher<PAYLOAD> publisher;
+        private final TestSubscriber<PAYLOAD> subscriber;
+
+        public Channel() {
+            publisher = new TestPublisher<>(HealthBase.TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+            subscriber = new TestSubscriber<>(HealthBase.TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+        }
+
+        /**
+         * Access this channel's publisher.
+         *
+         * @return this channel's test publisher
+         */
+        public TestPublisher<PAYLOAD> publisher() {
+            return publisher;
+        }
+
+        /**
+         * Access this channel's subscriber.
+         *
+         * @return this channel's test subscriber
+         */
+        public TestSubscriber<PAYLOAD> subscriber() {
+            return subscriber;
+        }
+
+        /**
+         * Trigger onSubscribe signal and block until received by subscriber.
+         *
+         * @return this channel
+         */
+        public Channel<PAYLOAD> ready() {
+            this.publisher().ready();
+            this.subscriber().awaitSubscription();
+            return this;
+        }
+
+        /**
+         * Trigger onComplete signal and block until received by subscriber.
+         *
+         * @return this channel
+         */
+        public Channel<PAYLOAD> complete() {
+            this.publisher().complete();
+            this.subscriber().awaitCompletion();
+            return this;
+        }
+
+        /**
+         * Trigger onError signal and block until received by subscriber.
+         *
+         * @return this channel
+         */
+        public Channel<PAYLOAD> fail() {
+            Exception exception = new Exception("BOOM!!!");
+            this.publisher().fail(exception);
+            Assert.assertEquals(exception.getMessage(), this.subscriber().awaitError().getMessage());
+            return this;
+        }
+
+        /**
+         * Trigger cancel signal and block until received by publisher.
+         *
+         * @return this channel
+         */
+        public Channel<PAYLOAD> cancel() {
+            this.subscriber().awaitSubscription().cancel();
+            this.publisher().awaitCancel();
+            return this;
+        }
+    }
+}

tck/src/main/java/org/eclipse/microprofile/reactive/messaging/tck/health/HealthAllExcludedTestBean.java

@@ -0,0 +1,48 @@
+/**
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * You may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.eclipse.microprofile.reactive.messaging.tck.health;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
+
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+import org.eclipse.microprofile.reactive.messaging.Outgoing;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+
+@ApplicationScoped
+public class HealthAllExcludedTestBean {
+
+    public static final String ALL_EXCLUDED_CHANNEL = "excluded-channel";
+
+    @Inject
+    private ChannelRegister channelRegisterBean;
+
+
+    @Incoming(ALL_EXCLUDED_CHANNEL)
+    public Subscriber<String> consumeInnerChannel() {
+        return channelRegisterBean.<String>get(ALL_EXCLUDED_CHANNEL).subscriber();
+    }
+
+    @Outgoing(ALL_EXCLUDED_CHANNEL)
+    public Publisher<String> produceInnerChannel() {
+        return channelRegisterBean.<String>get(ALL_EXCLUDED_CHANNEL).publisher();
+    }
+
+}