Use DirectProcessor in EventBus #1086

Lettuce now uses DirectProcessor as non-blocking event bus. DirectProcessor no longer blocks calling .next(…). Previously, EmitterProcessor used a blocking queue which blocked the caller if downstream consumers did not consume events in time.

src/main/java/io/lettuce/core/event/DefaultEventBus.java

@@ -15,8 +15,9 @@
  */
 package io.lettuce.core.event;
 
-import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
 import reactor.core.scheduler.Scheduler;
 
 /**
@@ -27,11 +28,13 @@
  */
 public class DefaultEventBus implements EventBus {
 
-    private final EmitterProcessor<Event> bus;
+    private final DirectProcessor<Event> bus;
+    private final FluxSink<Event> sink;
     private final Scheduler scheduler;
 
     public DefaultEventBus(Scheduler scheduler) {
-        this.bus = EmitterProcessor.create();
+        this.bus = DirectProcessor.create();
+        this.sink = bus.sink();
         this.scheduler = scheduler;
     }
 
@@ -42,8 +45,6 @@ public Flux<Event> get() {
 
     @Override
     public void publish(Event event) {
-        if (bus.hasDownstreams()) {
-            bus.onNext(event);
-        }
+        sink.next(event);
     }
 }