Repository: quarkusio/quarkus
Issue: Problem wiring up a SSE endpoint with an internal channel with the reactive messaging HTTP connector

### Describe the bug

I am using Quarkus 2.2.3 and OpenJDK 11.

The SSE endpoint and incoming channel looks like:
```
@Path("/hello")
public class ExampleResource {
    @Inject
    @Channel("sse-output")
    Multi<String> data;


    @GET
    @Produces(MediaType.SERVER_SENT_EVENTS)
    @RestSseElementType(MediaType.TEXT_PLAIN)
    public Multi<String> streamData() {
        return this.data;
    }
}

```


The producer is described below:

```
@ApplicationScoped
public class Producer {
    @Incoming("event")
    @Outgoing("sse-output")
    public String broadcast(final String time) {
        System.out.println("received event");

        return time.toUpperCase();
    }
}

```


The incoming "event" channel is defined in the application.resource file as:

mp.messaging.incoming.event.connector=quarkus-http
mp.messaging.incoming.event.path=/event
mp.messaging.incoming.event.method=POST
Dependencies in pom.xml
```
<dependencies>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-arc</artifactId>
</dependency>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-resteasy-reactive</artifactId>
</dependency>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-junit5</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.rest-assured</groupId>
<artifactId>rest-assured</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-reactive-messaging-http</artifactId>
</dependency>
</dependencies>
```

When I try to send a message using a HTTP POST, it throughs the following error:
```
2021-10-05 12:00:37,885 ERROR [io.qua.rea.htt.run.ReactiveHttpHandlerBean] (vert.x-eventloop-thread-2) Emitting message failed: java.lang.NullPointerException
at io.quarkus.reactivemessaging.http.runtime.ReactiveHttpHandlerBean.handleRequest(ReactiveHttpHandlerBean.java:61)
at io.quarkus.reactivemessaging.http.runtime.ReactiveHttpHandlerBean_Subclass.handleRequest$$superforward1(ReactiveHttpHandlerBean_Subclass.zig:549)
at io.quarkus.reactivemessaging.http.runtime.ReactiveHttpHandlerBean_Subclass$$function$$9.apply(ReactiveHttpHandlerBean_Subclass$$function$$9.zig:47)
at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:54)
at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.proceed(InvocationInterceptor.java:62)
at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.monitor(InvocationInterceptor.java:51)
at io.quarkus.arc.runtime.devconsole.InvocationInterceptor_Bean.intercept(InvocationInterceptor_Bean.zig:521)
at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:41)
at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:41)
at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
at io.quarkus.reactivemessaging.http.runtime.ReactiveHttpHandlerBean_Subclass.handleRequest(ReactiveHttpHandlerBean_Subclass.zig:1413)
at io.quarkus.reactivemessaging.http.runtime.ReactiveHandlerBeanBase.handle(ReactiveHandlerBeanBase.java:29)
at io.quarkus.reactivemessaging.http.runtime.ReactiveHttpHandlerBean_Subclass.handle$$superforward1(ReactiveHttpHandlerBean_Subclass.zig:530)
at io.quarkus.reactivemessaging.http.runtime.ReactiveHttpHandlerBean_Subclass$$function$$8.apply(ReactiveHttpHandlerBean_Subclass$$function$$8.zig:33)
at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:54)
at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.proceed(InvocationInterceptor.java:62)
at io.quarkus.arc.runtime.devconsole.InvocationInterceptor.monitor(InvocationInterceptor.java:51)
at io.quarkus.arc.runtime.devconsole.InvocationInterceptor_Bean.intercept(InvocationInterceptor_Bean.zig:521)
at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:41)
at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:41)
at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
at io.quarkus.reactivemessaging.http.runtime.ReactiveHttpHandlerBean_Subclass.handle(ReactiveHttpHandlerBean_Subclass.zig:1310)
at io.quarkus.reactivemessaging.http.runtime.ReactiveHttpHandler.handle(ReactiveHttpHandler.java:16)
at io.quarkus.reactivemessaging.http.runtime.ReactiveHttpHandler.handle(ReactiveHttpHandler.java:6)
at io.vertx.ext.web.impl.RouteState.handleContext(RouteState.java:1127)
at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:151)
at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:133)
at io.vertx.ext.web.handler.impl.BodyHandlerImpl$BHandler.doEnd(BodyHandlerImpl.java:313)
at io.vertx.ext.web.handler.impl.BodyHandlerImpl$BHandler.end(BodyHandlerImpl.java:290)
at io.vertx.ext.web.handler.impl.BodyHandlerImpl.lambda$handle$0(BodyHandlerImpl.java:86)
at io.vertx.core.impl.AbstractContext.dispatch(AbstractContext.java:100)
at io.vertx.core.impl.AbstractContext.dispatch(AbstractContext.java:63)
at io.vertx.core.http.impl.HttpEventHandler.handleEnd(HttpEventHandler.java:76)
at io.vertx.core.http.impl.Http1xServerRequest.onEnd(Http1xServerRequest.java:557)
at io.vertx.core.http.impl.Http1xServerRequest.lambda$pendingQueue$1(Http1xServerRequest.java:125)
at io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:240)
at io.vertx.core.streams.impl.InboundBuffer.drain(InboundBuffer.java:227)
at io.vertx.core.streams.impl.InboundBuffer.lambda$fetch$0(InboundBuffer.java:280)
at io.vertx.core.impl.AbstractContext.dispatch(AbstractContext.java:100)
at io.vertx.core.impl.AbstractContext.dispatch(AbstractContext.java:63)
at io.vertx.core.impl.EventLoopContext.lambda$runOnContext$0(EventLoopContext.java:38)
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:497)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:834)
```

The HTTP POST channel "event" works alone perfectly, however when I try to pass the event to the "see-out" channel I got the error above.

thank you if you can have a look at it.

Changed files:
extensions/resteasy-classic/rest-client/runtime/src/main/java/io/quarkus/restclient/runtime/RestClientBase.java
extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveHandlerBeanBase.java
extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/StreamConfigBase.java
extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveWebSocketHandlerBean.java
extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveHttpHandlerBean.java
extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/WebSocketStreamConfig.java
extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/HttpStreamConfig.java

diff --git a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveHandlerBeanBase.java b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveHandlerBeanBase.java
index 5a427728e8e..7b7fea04056 100644
--- a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveHandlerBeanBase.java
+++ b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveHandlerBeanBase.java
@@ -26,7 +26,7 @@ void handle(RoutingContext event) {
         if (bundle != null) {
             MultiEmitter<? super MessageType> emitter = bundle.emitter;
             StrictQueueSizeGuard guard = bundle.guard;
-            handleRequest(event, emitter, guard);
+            handleRequest(event, emitter, guard, bundle.path);
         } else {
             event.response().setStatusCode(404).end();
         }
@@ -41,6 +41,7 @@ private void addProcessor(ConfigType streamConfig) {
                 .<MessageType> emitter(bundle::setEmitter, BackPressureStrategy.BUFFER)
                 .onItem().invoke(guard::dequeue);
         bundle.setProcessor(processor);
+        bundle.setPath(streamConfig.path);
 
         Bundle<MessageType> previousProcessor = processors.put(key(streamConfig), bundle);
         if (previousProcessor != null) {
@@ -49,7 +50,7 @@ private void addProcessor(ConfigType streamConfig) {
     }
 
     protected abstract void handleRequest(RoutingContext event, MultiEmitter<? super MessageType> emitter,
-            StrictQueueSizeGuard guard);
+            StrictQueueSizeGuard guard, String path);
 
     protected abstract String description(ConfigType streamConfig);
 
@@ -63,6 +64,7 @@ protected abstract void handleRequest(RoutingContext event, MultiEmitter<? super
         private final StrictQueueSizeGuard guard;
         private Multi<MessageType> processor; // effectively final
         private MultiEmitter<? super MessageType> emitter; // effectively final
+        private String path;
 
         private Bundle(StrictQueueSizeGuard guard) {
             this.guard = guard;
@@ -79,5 +81,9 @@ public void setEmitter(MultiEmitter<? super MessageType> emitter) {
         public Multi<MessageType> getProcessor() {
             return processor;
         }
+
+        public void setPath(String path) {
+            this.path = path;
+        }
     }
 }
diff --git a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveHttpHandlerBean.java b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveHttpHandlerBean.java
index f6200f8b088..7641e4add5e 100644
--- a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveHttpHandlerBean.java
+++ b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveHttpHandlerBean.java
@@ -42,7 +42,7 @@ protected String key(HttpStreamConfig streamConfig) {
 
     @Override
     protected String key(RoutingContext context) {
-        return key(context.normalisedPath(), context.request().method());
+        return key(context.normalizedPath(), context.request().method());
     }
 
     @Override
@@ -52,8 +52,11 @@ protected String description(HttpStreamConfig streamConfig) {
 
     @Override
     protected void handleRequest(RoutingContext event, MultiEmitter<? super HttpMessage<?>> emitter,
-            StrictQueueSizeGuard guard) {
-        if (guard.prepareToEmit()) {
+            StrictQueueSizeGuard guard, String path) {
+        if (emitter == null) {
+            onUnexpectedError(event, null,
+                    "No consumer subscribed for messages sent to Reactive Messaging HTTP endpoint on path: " + path);
+        } else if (guard.prepareToEmit()) {
             try {
                 HttpMessage<Buffer> message = new HttpMessage<>(event.getBody(), new IncomingHttpMetadata(event.request()),
                         () -> event.response().setStatusCode(202).end(),
diff --git a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveWebSocketHandlerBean.java b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveWebSocketHandlerBean.java
index cb3b1983d82..86fdf27f3d8 100644
--- a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveWebSocketHandlerBean.java
+++ b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/ReactiveWebSocketHandlerBean.java
@@ -28,7 +28,7 @@ public class ReactiveWebSocketHandlerBean extends ReactiveHandlerBeanBase<WebSoc
 
     @Override
     protected void handleRequest(RoutingContext event, MultiEmitter<? super WebSocketMessage<?>> emitter,
-            StrictQueueSizeGuard guard) {
+            StrictQueueSizeGuard guard, String path) {
         event.request().toWebSocket(
                 webSocket -> {
                     if (webSocket.failed()) {
@@ -37,7 +37,11 @@ protected void handleRequest(RoutingContext event, MultiEmitter<? super WebSocke
                         ServerWebSocket serverWebSocket = webSocket.result();
                         serverWebSocket.handler(
                                 b -> {
-                                    if (guard.prepareToEmit()) {
+                                    if (emitter == null) {
+                                        onUnexpectedError(serverWebSocket, null,
+                                                "No consumer subscribed for messages sent to " +
+                                                        "Reactive Messaging WebSocket endpoint on path: " + path);
+                                    } else if (guard.prepareToEmit()) {
                                         try {
                                             emitter.emit(new WebSocketMessage<>(b,
                                                     () -> serverWebSocket.write(Buffer.buffer("ACK")),
@@ -67,7 +71,7 @@ protected String key(WebSocketStreamConfig config) {
 
     @Override
     protected String key(RoutingContext context) {
-        return context.normalisedPath();
+        return context.normalizedPath();
     }
 
     @Override
@@ -78,7 +82,7 @@ protected Collection<WebSocketStreamConfig> configs() {
     private void onUnexpectedError(ServerWebSocket serverWebSocket, Throwable error, String message) {
         log.error(message, error);
         // TODO some error message for the client? exception mapper would be best...
-        serverWebSocket.close((short) 3500, message);
+        serverWebSocket.close((short) 3500, "Unexpected error while processing the message");
     }
 
     Multi<WebSocketMessage<?>> getProcessor(String path) {
diff --git a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/HttpStreamConfig.java b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/HttpStreamConfig.java
index 127302d4102..b500b21a227 100644
--- a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/HttpStreamConfig.java
+++ b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/HttpStreamConfig.java
@@ -6,11 +6,9 @@
 
 public class HttpStreamConfig extends StreamConfigBase {
     public final HttpMethod method;
-    public final String path;
 
     public HttpStreamConfig(String path, String method, String name, int bufferSize) {
-        super(bufferSize);
-        this.path = path;
+        super(bufferSize, path);
         this.method = toHttpMethod(method, name);
     }
 
diff --git a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/StreamConfigBase.java b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/StreamConfigBase.java
index caed4a514a8..0174103016a 100644
--- a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/StreamConfigBase.java
+++ b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/StreamConfigBase.java
@@ -2,8 +2,10 @@
 
 public class StreamConfigBase {
     public final int bufferSize;
+    public final String path;
 
-    public StreamConfigBase(int bufferSize) {
+    public StreamConfigBase(int bufferSize, String path) {
+        this.path = path;
         this.bufferSize = bufferSize;
     }
 }
diff --git a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/WebSocketStreamConfig.java b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/WebSocketStreamConfig.java
index c2ff5316ae0..5d1eea4357b 100644
--- a/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/WebSocketStreamConfig.java
+++ b/extensions/reactive-messaging-http/runtime/src/main/java/io/quarkus/reactivemessaging/http/runtime/config/WebSocketStreamConfig.java
@@ -1,11 +1,8 @@
 package io.quarkus.reactivemessaging.http.runtime.config;
 
 public class WebSocketStreamConfig extends StreamConfigBase {
-    public final String path;
-
     public WebSocketStreamConfig(String path, int bufferSize) {
-        super(bufferSize);
-        this.path = path;
+        super(bufferSize, path);
     }
 
     public String path() {
diff --git a/extensions/resteasy-classic/rest-client/runtime/src/main/java/io/quarkus/restclient/runtime/RestClientBase.java b/extensions/resteasy-classic/rest-client/runtime/src/main/java/io/quarkus/restclient/runtime/RestClientBase.java
index b41ee0ec82a..9ef6e41090a 100644
--- a/extensions/resteasy-classic/rest-client/runtime/src/main/java/io/quarkus/restclient/runtime/RestClientBase.java
+++ b/extensions/resteasy-classic/rest-client/runtime/src/main/java/io/quarkus/restclient/runtime/RestClientBase.java
@@ -318,7 +318,7 @@ private static RestClientsConfig getConfigRoot() {
         InstanceHandle<RestClientsConfig> configHandle = Arc.container()
                 .instance(RestClientsConfig.class);
         if (!configHandle.isAvailable()) {
-            throw new IllegalStateException("Unable to find the RestClientConfigRootProvider");
+            throw new IllegalStateException("Unable to find the RestClientConfigs");
         }
         return configHandle.get();
     }