EventBus ThreadPool sometimes executes in producer thread

[connollyst]

Hi again,

Another v1 to v2 migration issue here.

I'm using eventBus.notify(event) where my eventBus is using a ThreadPoolExecutorDispatcher (Spring Configuration at the end). I see that my events are being consumed and processed by the 5 threads in my thread pool. However, every so often the consumer is being executed in the event origin thread.

This is not the expected behavior and seems to be no to v2. What's going on here?

I think this might be related to this unresolved conversation from last year: "THREAD POOL is not working with different threads"

I can't figure out the take away from that discussion though.

To be honest we can't figure out what's going on with Reactor. This v1 -> v2 rewrite seems to have made a mess of thing. Changing the APIs without providing the documentation isn't really how things should be done. One of your main portals is that Spring promoted this project as an asynchronous event bus.. but that example code no longer works, eek!

I hate to say it but our company will surely be moving away from Reactor going forward. Still, perhaps this is a bug that can be fixed, or a "feature" that should be documented. Hopefully this issue report helps someone.

Good luck.

private static final int THREAD_POOL_SIZE = 5;
private static final int BACKLOG_SIZE = 1024;
private static final String THREAD_NAME = "my-reactor";

@Bean
public EventBus reactor(Environment env) {
    LOG.debug("Creating reactor..");
    return new EventBusSpec()
            .env(env)
                .dispatcher(reactorDispatcher())
                .consumerNotFoundHandler(notificationKey -> {
                    if(notificationKey != null) {
                        String key = notificationKey.toString();
                        if(key.startsWith(ExternalDataEvent.SELECTOR)) {
                            LOG.error("No consumer found for {}", key);
                        }
                    }
                })
                .broadcastEventRouting()
                .uncaughtErrorHandler(error -> LOG.error(
                        "An error occurred while processing event: {}", error.getMessage(), error))
                .get();
}

@Bean
public Dispatcher reactorDispatcher() {
    LOG.debug("Creating thread pool dispatcher with thread prefix '{}'", THREAD_NAME);
    return new ThreadPoolExecutorDispatcher(THREAD_POOL_SIZE, BACKLOG_SIZE, THREAD_NAME);
}

[smaldini]

Moved to http://github.com/reactor/reactor-bus