如何将反应流连接到quarkus/smallrye中的AMQP代理

Question

我正在尝试将我的Artimis-MQ客户端迁移到quarkus微服务。当试图发送消息时，我总是会收到一个“流未连接”错误。

我试图遵循答案中的建议（使用Microprofile-reactive-messaging）：Quarkus与ActiveMQ？

dependencies {
    // ...
    implementation enforcedPlatform("io.quarkus:quarkus-bom:0.15.0")
    implementation 'io.quarkus:quarkus-resteasy'
    implementation 'io.quarkus:quarkus-resteasy-jsonb'
    implementation 'io.quarkus:quarkus-smallrye-metrics'
    implementation 'io.quarkus:quarkus-smallrye-health'
    implementation 'io.quarkus:quarkus-smallrye-reactive-messaging'
    implementation 'io.quarkus:quarkus-vertx'

    implementation 'io.smallrye.reactive:smallrye-reactive-messaging-amqp:0.0.8'
}

@Path("/send")
public class MessageResource {
    @Inject
    @Stream("emitter-topic")
    Emitter<String> topic;

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    public String send(@QueryParam("msg") final String msg) {
        final String message = Objects.requireNonNullElse(msg, "").isBlank() ? "no message" : msg;
        topic.send(message);
        return "sent: " + message;
    }
}

smallrye.messaging.source.emitter-topic.type=io.smallrye.reactive.messaging.amqp.Amqp
smallrye.messaging.source.emitter-topic.address=test-amqp
smallrye.messaging.source.emitter-topic.containerId=test-clientid
smallrye.messaging.source.emitter-topic.host=localhost
smallrye.messaging.source.emitter-topic.port=5672

我不断地看到“非法状态例外”。我可以从日志中看出，smallrye找到了amqp连接器，但从未实际初始化连接。

2019-06-02 12:19:50,055 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Deployment done... start processing
2019-06-02 12:19:50,101 INFO  [io.sma.rea.mes.imp.ConfiguredStreamFactory] (main) Found incoming connectors: [class io.smallrye.reactive.messaging.amqp.Amqp]
2019-06-02 12:19:50,102 INFO  [io.sma.rea.mes.imp.ConfiguredStreamFactory] (main) Found outgoing connectors: [class io.smallrye.reactive.messaging.amqp.Amqp]
2019-06-02 12:19:50,103 INFO  [io.sma.rea.mes.imp.ConfiguredStreamFactory] (main) Stream manager initializing...
2019-06-02 12:19:50,106 INFO  [io.sma.rea.mes.imp.LegacyConfiguredStreamFactory] (main) Stream manager initializing...
2019-06-02 12:19:50,125 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Initializing mediators
2019-06-02 12:19:50,127 INFO  [io.sma.rea.mes.ext.MediatorManager] (main) Connecting mediators
2019-06-02 12:19:50,136 INFO  [io.quarkus] (main) Quarkus 0.15.0 started in 1.487s. Listening on: http://[::]:8080
2019-06-02 12:19:50,137 INFO  [io.quarkus] (main) Installed features: [cdi, resteasy, resteasy-jsonb, smallrye-health, smallrye-metrics, smallrye-reactive-messaging, smallrye-reactive-streams-operators, vertx]
2019-06-02 12:20:01,964 ERROR [io.und.request] (executor-thread-1) UT005023: Exception handling request to /send: org.jboss.resteasy.spi.UnhandledException: java.lang.IllegalStateException: Stream not yet connected
        at org.jboss.resteasy.core.ExceptionHandler.handleApplicationException(ExceptionHandler.java:106)
        at org.jboss.resteasy.core.ExceptionHandler.handleException(ExceptionHandler.java:372)
        at org.jboss.resteasy.core.SynchronousDispatcher.writeException(SynchronousDispatcher.java:209)
        at org.jboss.resteasy.core.SynchronousDispatcher.invoke(SynchronousDispatcher.java:496)
        at org.jboss.resteasy.core.SynchronousDispatcher.lambda$invoke$4(SynchronousDispatcher.java:252)
        at org.jboss.resteasy.core.SynchronousDispatcher.lambda$preprocess$0(SynchronousDispatcher.java:153)
        at org.jboss.resteasy.core.interception.jaxrs.PreMatchContainerRequestContext.filter(PreMatchContainerRequestContext.java:362)
        at org.jboss.resteasy.core.SynchronousDispatcher.preprocess(SynchronousDispatcher.java:156)
        at org.jboss.resteasy.core.SynchronousDispatcher.invoke(SynchronousDispatcher.java:238)
        at org.jboss.resteasy.plugins.server.servlet.ServletContainerDispatcher.service(ServletContainerDispatcher.java:234)
        at io.quarkus.resteasy.runtime.ResteasyFilter$ResteasyResponseWrapper.sendError(ResteasyFilter.java:72)
        at io.undertow.servlet.handlers.DefaultServlet.doGet(DefaultServlet.java:175)
        at javax.servlet.http.HttpServlet.service(HttpServlet.java:686)

Answer 1

好吧，我解决了我的问题。在application.properties中，我反向使用了source和sink。将发射器-主题描述为接收器而不是源解决了这个问题。