java - 如何将 react 流连接到 quarkus/smallrye 中的 AMQP 代理

Question

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

我尝试遵循答案中的建议(使用 microprofile-reactive-messaging):Quarkus with ActiveMQ?

在我的 build.gradle 中:

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'
}

示例休息端点，将消息转发到 AMQP

@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;
    }
}

在src/main/resources/application.properties中:

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向后。将emitter-topic描述为接收器，而不是源解决了问题。