quarkiverse / quarkus-reactive-messaging-http

Connect to HTTP or Web Socket and expose HTTP or Web Socket endpoints for Reactive Messaging

Quarkus Extension for Reactive Messaging with HTTP

This project is a Quarkus extension that lets you utilize SmallRye Reactive Messaging to consume and produce HTTP and WebSocket messages

This extension was previously a part of the core Quarkus repository.

⚠️ Version 2.x.x of this extension (main branch) supports Quarkus 3, and version 1.x.x (quarkus2 branch) supports Quarkus 2.

Documentation

Documentation for the project can be found on the Quarkiverse Documentation website.

Contributing

You can find the basic info, Quarkiverse policies and conventions in the Quarkiverse wiki.

Info building extensions can be found in Building My First Extension guide.

Other useful articles related to Quarkus extension development can be found under the Writing Extensions guide category on the Quarkus.io website.

Documentation

The documentation for this extension should be maintained as part of this repository and it is stored in the docs/ directory.

The layout should follow the Antora's Standard File and Directory Set.

Once the docs are ready to be published, please open a PR including this repository in the Quarkiverse Docs Antora playbook. See an example here.

Contributors ✨

Thanks goes to these wonderful people (emoji key):

Michał Szynkiewicz 💻 🚧

This project follows the all-contributors specification. Contributions of any kind welcome!

quarkus-reactive-messaging-http's Issues

Support binary form of Cloud Events

We found with @fjtirado that when we send a cloud event in a binary form (with extension attributes and other metadata in ce headers), this data is not available to the application. We have to use structured form where everything is in the body of the request.

It would be good if the http connector supported also binary form of cloud events for better flexibility.

All contributors

ce_time header is not following RFC-3339 when seconds are 0

HttpCloudEventHelper.getCloudEventHeaders is using ZonedDateTime.toString() method for the ce_time HTTP header. In case seconds are 0 the time doesn't contain any :00 seconds part. This is not conforming to https://datatracker.ietf.org/doc/html/rfc3339#section-5.6 where seconds are mandatory.

Upgrade release.yml to using Environment Files.

The set-output command is deprecated and will be disabled soon. We need to upgrade https://github.com/quarkiverse/quarkus-reactive-messaging-http/blob/main/.github/workflows/release.yml to using Environment Files. For more information see: https://github.blog/changelog/2022-10-11-github-actions-deprecating-save-state-and-set-output-commands/

Reactive Messaging WebSockets connector - exception thrown on close

Moved from quarkusio/quarkus#19662

Describe the bug

The following exception is thrown on application shutdown when using the quarkus websocket connector for reactive messaging:

2021-08-25 21:40:56,651 ERROR [io.qua.ver.cor.run.VertxCoreRecorder] (vert.x-eventloop-thread-40) Uncaught exception received by Vert.x: java.lang.IllegalStateException
        at io.vertx.core.impl.CloseFuture.add(CloseFuture.java:51)
        at io.vertx.core.impl.ContextInternal.addCloseHook(ContextInternal.java:294)
        at io.vertx.core.impl.VertxImpl.scheduleTimeout(VertxImpl.java:524)
        at io.vertx.core.impl.VertxImpl.setTimer(VertxImpl.java:372)
        at io.vertx.core.http.impl.WebSocketImplBase.initiateConnectionCloseTimeout(WebSocketImplBase.java:567)
        at io.vertx.core.http.impl.ServerWebSocketImpl.lambda$close$0(ServerWebSocketImpl.java:149)
        at io.vertx.core.impl.future.FutureImpl$3.onSuccess(FutureImpl.java:124)
        at io.vertx.core.impl.future.FutureBase.lambda$emitSuccess$0(FutureBase.java:54)
        at io.vertx.core.impl.EventLoopContext.execute(EventLoopContext.java:80)
        at io.vertx.core.impl.DuplicatedContext.execute(DuplicatedContext.java:172)
        at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:51)
        at io.vertx.core.impl.future.FutureImpl.addListener(FutureImpl.java:164)
        at io.vertx.core.impl.future.PromiseImpl.addListener(PromiseImpl.java:23)
        at io.vertx.core.impl.future.FutureImpl.onComplete(FutureImpl.java:132)
        at io.vertx.core.impl.future.PromiseImpl.onComplete(PromiseImpl.java:23)
        at io.vertx.core.http.impl.ServerWebSocketImpl.close(ServerWebSocketImpl.java:145)
        at io.vertx.core.http.impl.WebSocketImplBase.close(WebSocketImplBase.java:143)
        at io.vertx.core.http.impl.Http1xConnectionBase.close(Http1xConnectionBase.java:95)
        at io.vertx.core.http.impl.Http1xServerConnection.close(Http1xServerConnection.java:75)
        at io.vertx.core.net.impl.ConnectionBase.close(ConnectionBase.java:130)
        at io.vertx.core.net.impl.VertxHandler.close(VertxHandler.java:159)
        at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:622)
        at io.netty.channel.AbstractChannelHandlerContext.access$1200(AbstractChannelHandlerContext.java:61)
        at io.netty.channel.AbstractChannelHandlerContext$11.run(AbstractChannelHandlerContext.java:611)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
        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)

2021-08-25 21:40:56,651 ERROR [io.qua.ver.cor.run.VertxCoreRecorder] (vert.x-eventloop-thread-36) Uncaught exception received by Vert.x: java.lang.IllegalStateException
        at io.vertx.core.impl.CloseFuture.add(CloseFuture.java:51)
        at io.vertx.core.impl.ContextInternal.addCloseHook(ContextInternal.java:294)
        at io.vertx.core.impl.VertxImpl.scheduleTimeout(VertxImpl.java:524)
        at io.vertx.core.impl.VertxImpl.setTimer(VertxImpl.java:372)
        at io.vertx.core.http.impl.WebSocketImplBase.initiateConnectionCloseTimeout(WebSocketImplBase.java:567)
        at io.vertx.core.http.impl.ServerWebSocketImpl.lambda$close$0(ServerWebSocketImpl.java:149)
        at io.vertx.core.impl.future.FutureImpl$3.onSuccess(FutureImpl.java:124)
        at io.vertx.core.impl.future.FutureBase.lambda$emitSuccess$0(FutureBase.java:54)
        at io.vertx.core.impl.EventLoopContext.execute(EventLoopContext.java:80)
        at io.vertx.core.impl.DuplicatedContext.execute(DuplicatedContext.java:172)
        at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:51)
        at io.vertx.core.impl.future.FutureImpl.addListener(FutureImpl.java:164)
        at io.vertx.core.impl.future.PromiseImpl.addListener(PromiseImpl.java:23)
        at io.vertx.core.impl.future.FutureImpl.onComplete(FutureImpl.java:132)
        at io.vertx.core.impl.future.PromiseImpl.onComplete(PromiseImpl.java:23)
        at io.vertx.core.http.impl.ServerWebSocketImpl.close(ServerWebSocketImpl.java:145)
        at io.vertx.core.http.impl.WebSocketImplBase.close(WebSocketImplBase.java:143)
        at io.vertx.core.http.impl.Http1xConnectionBase.close(Http1xConnectionBase.java:95)
        at io.vertx.core.http.impl.Http1xServerConnection.close(Http1xServerConnection.java:75)
        at io.vertx.core.net.impl.ConnectionBase.close(ConnectionBase.java:130)
        at io.vertx.core.net.impl.VertxHandler.close(VertxHandler.java:159)
        at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:622)
        at io.netty.channel.AbstractChannelHandlerContext.access$1200(AbstractChannelHandlerContext.java:61)
        at io.netty.channel.AbstractChannelHandlerContext$11.run(AbstractChannelHandlerContext.java:611)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
        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)

Expected behavior

No response

Actual behavior

No response

How to Reproduce?

No response

Output of uname -a or ver

No response

Output of java -version

No response

GraalVM version (if different from Java)

No response

Quarkus version or git rev

No response

Build tool (ie. output of mvnw --version or gradlew --version)

No response

Additional information

No response

Reactive HTTP and WebSocket connector enhancements

• add support for STOMP or some other protocol that allows acking/nacking by the remote websocket endpoint
• support text messages for websockets
• add content type headers in http serializers
• websocket source exception mapper
• improve conversion error reporting (ATM we have 500 instead of 422 in such case)
• give a possibility to read response for sent out messages

moved from Quarkus

"Unable to find matching target resource method" in native mode for incoming channel

Working on use of http reactive messaging and all works properly in JVM mode but it then results in Unable to find matching target resource method when running in native mode.

I have additional custom ConfigSouce that carries information about the connector, path etc. It seems to be found at runtime but somehow does not register something, I guess on the reactive routes level but no idea how to trace it.

When same values are explicitly added to application.properties then the route is registered.

Any hints how to troubleshoot this case?

HTTP and CloudEvent headers cannot be set outgoing

When both http outgoing headers and cloud event meta data are defined only the HTTP headers are sent to the outgoing channel destination.

Incompatibility in Jandex version 3 and 2.x

Jandex version 3 will be available in the next Quarkus release ( 2.14 ), in there methodInfo.arguments will have a different return type, which end up breaking during the Kogito compilation when aligning with Quarkus.
To keep compatibility across different versions of Jandex, methodInfo.args array can be used.
Kogito stacktrace below.

[2022-09-20T06:40:21.929Z] [ERROR] Failed to execute goal io.quarkus:quarkus-maven-plugin:999-SNAPSHOT:build (default) on project kogito-quarkus-serverless-workflow-integration-test: Failed to build quarkus application: io.quarkus.builder.BuildException: Build failure: Build failed due to errors

[2022-09-20T06:40:21.929Z] [ERROR] [error]: Build step io.quarkus.reactivemessaging.http.deployment.ReactiveHttpProcessor#registerMessagePayloadClassesForReflection threw an exception: java.lang.ClassCastException: class org.jboss.jandex.MethodParameterInfo cannot be cast to class org.jboss.jandex.Type (org.jboss.jandex.MethodParameterInfo and org.jboss.jandex.Type are in unnamed module of loader io.quarkus.bootstrap.classloading.QuarkusClassLoader @14458063)

[2022-09-20T06:40:21.929Z] [ERROR] at io.quarkus.reactivemessaging.http.deployment.ReactiveHttpProcessor.registerMessagePayloadClassesForReflection(ReactiveHttpProcessor.java:143)

List configuration properties on the pages

move issues from quarkus core repo and remove the extension there

Add support to Quarkus 3

Quarkus migration tips: https://github.com/quarkiverse/quarkiverse/wiki/Migrating-to-Quarkus-3.x

provide support for broadcast on incoming channel

Currently there is no way to use broadcast configuration of the incoming channel to invoke all subscribers. It is supported for other reactive messaging connectors and for http it makes it even better as single call to endpoint can invoke multiple subscribers efficiently.

move quickstart and guide from quarkusio org

Use -DnoCode instead of -DnoExamples and add resteasy-reactive extension in the guides

See quarkusio/quarkus#24948 (comment)

mvn io.quarkus.platform:quarkus-maven-plugin:2.8.0.Final:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=reactive-messaging-http-quickstart \
    -Dextensions="reactive-messaging-http,resteasy-reactive" \
    -DnoCode

move native test from quarkus core repository

How can I set Connection Timeout & Read Timeout ?

I was looking for the settings for connect and read timeout. I checked the code that HttpSink class using WebClientOptions but without timeout option. What is the most convenient way to set read & connect timeout for your library? Should I use SmallRye Fault Tolerance instead of the setting those attributes in it ?

Best Regards