Throttle unread status changes.

Reduces unread status changes to publish to the eventbus
at most once every 200ms per window. These events are
handled synchronously on the EDT so cause massive
performance issues.

When connecting to several bouncers with huge backbuffers
this makes the client usable in seconds rather than tens of minutes.

Internally this uses RX to handle the throttling, because it's
super easy. We might eventually want to expose that externally
instead of using an event bus, but I'm not sure.

build.gradle

@@ -61,7 +61,8 @@ dependencies {
     bundle group: 'com.getsentry.raven', name: 'raven', version: '7.8.0'
     bundle group: 'com.google.guava', name:'guava', version: '19.0'
     bundle group: 'net.engio', name: 'mbassador', version: '1.3.0'
-    bundle group: 'com.google.code.gson', name: 'gson', 'version': '2.5'
+    bundle group: 'com.google.code.gson', name: 'gson', version: '2.5'
+    bundle group: 'io.reactivex.rxjava2', name: 'rxjava', version: '2.1.12'
 
     bundle group: 'com.dmdirc', name: 'com.dmdirc.config.binding', version: '+', changing: true
     bundle group: 'com.dmdirc', name: 'com.dmdirc.config.provider', version: '+', changing: true

src/main/java/com/dmdirc/ui/messages/UnreadStatusManagerImpl.java

@@ -30,17 +30,27 @@ import com.dmdirc.interfaces.WindowModel;
 import com.dmdirc.util.colours.Colour;
 
 import java.util.Optional;
-
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.Emitter;
+import io.reactivex.Observable;
+import io.reactivex.ObservableEmitter;
+import io.reactivex.Observer;
+import io.reactivex.functions.Consumer;
+import io.reactivex.subjects.PublishSubject;
+import io.reactivex.subjects.ReplaySubject;
+import io.reactivex.subjects.Subject;
 import net.engio.mbassy.listener.Handler;
+import org.reactivestreams.Publisher;
 
 /**
  * Tracks unread messages and other notifications.
  */
 public class UnreadStatusManagerImpl implements UnreadStatusManager {
 
-    private final EventBus eventBus;
     private final WindowModel container;
     private final ColourManager colourManager;
+    private final Subject<UnreadStatusChangedEvent> eventSubject;
 
     private int unreadLines;
     private Optional<Colour> notificationColour = Optional.empty();
@@ -51,8 +61,12 @@ public class UnreadStatusManagerImpl implements UnreadStatusManager {
 
     public UnreadStatusManagerImpl(final WindowModel container) {
         this.container = container;
-        this.eventBus = container.getEventBus();
         this.colourManager = new ColourManagerImpl(container.getConfigManager());
+        this.eventSubject = PublishSubject.create();
+
+        eventSubject
+                .throttleLast(200, TimeUnit.MILLISECONDS)
+                .subscribe(unreadStatusChangedEvent -> container.getEventBus().publish(unreadStatusChangedEvent));
     }
 
     @Handler
@@ -177,7 +191,7 @@ public class UnreadStatusManagerImpl implements UnreadStatusManager {
     }
 
     private void publishChangedEvent() {
-        eventBus.publishAsync(new UnreadStatusChangedEvent(container, this, notificationColour,
+        eventSubject.onNext(new UnreadStatusChangedEvent(container, this, notificationColour,
                 unreadLines));
     }