Как подключить реактивный поток к брокеру AMQP в quarkus / smallrye

Я пытаюсь перенести свои клиенты Artimis-MQ на микросервисы quarkus. Я постоянно получаю сообщение об ошибке «Поток не подключен» при попытке отправить сообщение.

Я попытался следовать предложениям в ответе (используя микропрофильный реактивный обмен сообщениями): Quarkus с 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;
    }
}

in 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)

java quarkus smallrye-reactive-messaging
person Joe    schedule 02.06.2019    source источник

Ответы (1)


arrow_upward
1
arrow_downward

Хорошо, я понял свою проблему. В application.properties у меня были source и sink наоборот. Описание emitter-topic как приемника, а не источника, решило проблему.

person Joe    schedule 02.06.2019