CODEWITHSUNDEEP
spring-bootvert.xvertx-eventbus
greymatter
greymatter

Reputation: 990

eventBus().consumer() never gets called

I have added vert.x to a Spring Boot application by adding these dependencies to build.gradle:

    compile "io.vertx:vertx-core:3.8.1"
    compile "io.vertx:vertx-lang-groovy:3.8.1"

I want to use the vert.x EventBus to implement reactive code within the single JVM application (no Verticles).

I have verified that outbound interceptors work and that SharedData works. However, there is no sign of inbound interceptors or consumers ever being invoked.

I suspect that there is something I am overlooking with configuring vert.x or that embedding vert.x inside Spring Boot is somehow preventing receipt of the inbound messages.

        Vertx vertx = Vertx.vertx();

        vertx.eventBus().addInboundInterceptor(msg -> {
            log.debug("abc inbound "+msg);
        });

        vertx.eventBus().addOutboundInterceptor(msg -> {
            log.debug("abc outbound "+msg);
        });

        vertx.eventBus().<String>consumer("abc", (Message<String> msg) -> {
            log.debug("abc handler");
        });

        vertx.eventBus().<String>localConsumer("localabc", (Message<String> msg) -> {
            log.debug("local abc handler");
        });

        vertx.eventBus().consumer("abc", msg -> {
            log.debug("abc handler 2");
        });

        vertx.eventBus().localConsumer("localabc", msg -> {
            log.debug("local abc handler 2");
        });

        MessageConsumer<String> consumer1 = vertx.eventBus().consumer("abc");
        consumer1.handler(msg -> {
            log.debug("abc handler 3");
        });

        MessageConsumer<String> consumer2 = vertx.eventBus().localConsumer("localabc");
        consumer2.handler(msg -> {
            log.debug("local abc handler 3");
        });

        LocalMap<String, String> localMap = vertx.sharedData().getLocalMap("abc");
        localMap.put("abc", "abc map");

        vertx.eventBus().publish("abc", "test", new DeliveryOptions().setLocalOnly(true));
        vertx.eventBus().publish("localabc", "localtest", new DeliveryOptions().setLocalOnly(true));
        //LocalMap<String, String> localMap = vertx.sharedData().getLocalMap("abc");
        log.debug("abc map contains "+localMap.get("abc"));

Here is the output. There are no errors of any kind.

abc outbound io.vertx.core.eventbus.impl.EventBusImpl$OutboundDeliveryContext@3bab9a17
abc outbound io.vertx.core.eventbus.impl.EventBusImpl$OutboundDeliveryContext@54887f7e
abc map contains abc map

Upvotes: 0

Views: 612

Answers (1)

Alexey Soshin
Alexey Soshin

Reputation: 17691

You're using outboundInterceptor without next()

So it acts like a filter. It catches all your messages, and never forwards them to consumers.

You can just use:

vertx.eventBus().addOutboundInterceptor(msg -> {
    log.debug("abc outbound "+msg);
    msg.next();
});

Upvotes: 1

Related Questions