Add support for smallrye-reactive-messaging-camel

[ddcprg]

Currently it seems only smallrye-reactive-messaging-kafka is supported. Adding support for reactive Camel would be useful in my opinion. Is this is in the roadmap?

I'd would like to contribute this but honestly I don't know where to start.

[cescoffier]

The first step is to try with the upstream artifact and see if it works out of the box.

Then, we would need to depends on quarkus-camel instead of the upstream camel.

[ddcprg]

I've got a working example but as soon as I try to use quarkus-camel-core the application fails. I noticed that this artifact has a dependency on Camel 3 (which is still under development). I think this is causing the exception since smallrye-reactive-messaging-camel depends on previous version 2.23.1.

I'll try to clean up the code and push it a public repo in my account.

[ddcprg]

I have created this repo https://github.com/ddcprg/camel-stream-test which is in fact using quarkus-camel-core

At the moment I'm not using the Smallrye annotations to define the stream and instead I'm just creating the reactive stream programmatically with a Camel route subscriber.

This example uses Camel 3.0.0-M1 and I manage to get it working with smallrye-reactive-messaging-camel version 0.0.4 compiled to use this version of Camel. If I try to use a later version smallrye-reactive-messaging-camel (I think Quarkus depends on 0.0.5 for Kafka actually) then I get this exception on startup:

19:05:02,568 ERROR [io.qua.dev.DevModeMain] Failed to start quarkus: java.lang.RuntimeException: org.jboss.builder.BuildException: Build failure: Build failed due to errors
        [error]: Build step io.quarkus.arc.deployment.ArcAnnotationProcessor#build threw an exception: java.lang.IllegalStateException: Failed to index: io.smallrye.reactive.messaging.spi.IncomingConnectorFactory
        at io.quarkus.runner.RuntimeRunner.run(RuntimeRunner.java:134)
        at io.quarkus.dev.DevModeMain.doStart(DevModeMain.java:105)
        at io.quarkus.dev.DevModeMain.main(DevModeMain.java:66)
Caused by: org.jboss.builder.BuildException: Build failure: Build failed due to errors
        [error]: Build step io.quarkus.arc.deployment.ArcAnnotationProcessor#build threw an exception: java.lang.IllegalStateException: Failed to index: io.smallrye.reactive.messaging.spi.IncomingConnectorFactory
        at org.jboss.builder.Execution.run(Execution.java:123)
        at org.jboss.builder.BuildExecutionBuilder.execute(BuildExecutionBuilder.java:136)
        at io.quarkus.deployment.QuarkusAugmentor.run(QuarkusAugmentor.java:110)
        at io.quarkus.runner.RuntimeRunner.run(RuntimeRunner.java:99)
        ... 2 more
Caused by: java.lang.IllegalStateException: Failed to index: io.smallrye.reactive.messaging.spi.IncomingConnectorFactory
        at io.quarkus.arc.processor.BeanArchives.index(BeanArchives.java:233)
        at io.quarkus.arc.processor.BeanArchives$IndexWrapper.lambda$getClassByName$0(BeanArchives.java:105)
        at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1660)
        at io.quarkus.arc.processor.BeanArchives$IndexWrapper.getClassByName(BeanArchives.java:102)
        at io.quarkus.arc.processor.Types.getTypeClosure(Types.java:216)
        at io.quarkus.arc.processor.Beans.createClassBean(Beans.java:55)
        at io.quarkus.arc.processor.BeanDeployment.findBeans(BeanDeployment.java:615)
        at io.quarkus.arc.processor.BeanDeployment.<init>(BeanDeployment.java:136)
        at io.quarkus.arc.processor.BeanProcessor.process(BeanProcessor.java:150)
        at io.quarkus.arc.deployment.ArcAnnotationProcessor.build(ArcAnnotationProcessor.java:237)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at io.quarkus.deployment.ExtensionLoader$1.execute(ExtensionLoader.java:506)
        at org.jboss.builder.BuildContext.run(BuildContext.java:413)
        at org.jboss.threads.ContextClassLoaderSavingRunnable.run(ContextClassLoaderSavingRunnable.java:35)
        at org.jboss.threads.EnhancedQueueExecutor.safeRun(EnhancedQueueExecutor.java:1998)
        at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.doRunTask(EnhancedQueueExecutor.java:1525)
        at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1416)
        at java.lang.Thread.run(Thread.java:748)
        at org.jboss.threads.JBossThread.run(JBossThread.java:479)
Caused by: java.io.IOException: Stream closed
        at java.io.BufferedInputStream.getInIfOpen(BufferedInputStream.java:159)
        at java.io.BufferedInputStream.fill(BufferedInputStream.java:246)
        at java.io.BufferedInputStream.read1(BufferedInputStream.java:286)
        at java.io.BufferedInputStream.read(BufferedInputStream.java:345)
        at java.io.DataInputStream.readFully(DataInputStream.java:195)
        at java.io.DataInputStream.readFully(DataInputStream.java:169)
        at org.jboss.jandex.Indexer.verifyMagic(Indexer.java:1288)
        at org.jboss.jandex.Indexer.index(Indexer.java:1587)
        at io.quarkus.arc.processor.BeanArchives.index(BeanArchives.java:231)
        ... 21 more

I'll try to dig a bit more into this issue later.

[DuncanDoyle]

This would be super useful for Kogito. In Kogito we generate SmallRye Reactive Messaging Consumers and Producers from BPMN2 message events (e.g. Start Message Event, End Message Event). Currently we mostly use the Kafka connector for our use-cases. Being able to use the Camel Connector would open loads of new integration possibilities for us!

[cescoffier]

The connector is using Camel-CDI which does not seem to be available in quarkus-camel. @lburgazzoli can you confirm?

[cescoffier]

After discussing it with @lburgazzoli, we should implement the extension in camel quarkus.

[lburgazzoli]

Opened an issue on camel-quarkus: apache/camel-quarkus#1469

[cescoffier]

Done and documented: https://camel.apache.org/camel-quarkus/latest/reference/extensions/smallrye-reactive-messaging.html

@lburgazzoli please thank the contributor for me!