From cac86d1f77cd3850d68fdc7b5ef8dba9f32480a1 Mon Sep 17 00:00:00 2001
From: Jon Chambers <jon@signal.org>
Date: Sun, 10 Jan 2021 14:53:59 -0500
Subject: [PATCH] Standardize toplogy event handling strategy.

---
 .../textsecuregcm/push/ClientPresenceManager.java            | 5 +----
 .../whispersystems/textsecuregcm/storage/MessagesCache.java  | 5 +----
 2 files changed, 2 insertions(+), 8 deletions(-)

diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/push/ClientPresenceManager.java b/service/src/main/java/org/whispersystems/textsecuregcm/push/ClientPresenceManager.java
index 772bb98ba..5a7b85c65 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/push/ClientPresenceManager.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/push/ClientPresenceManager.java
@@ -107,10 +107,7 @@ public class ClientPresenceManager extends RedisClusterPubSubAdapter<String, Str
             connection.addListener(this);
             connection.getResources().eventBus().get()
                                                 .filter(event -> event instanceof ClusterTopologyChangedEvent)
-                                                .handle((event, sink) -> {
-                                                    resubscribeAll();
-                                                    sink.next(event);
-                                                });
+                                                .subscribe(event -> resubscribeAll());
 
             final String presenceChannel = getManagerPresenceChannel(managerId);
             final int    slot            = SlotHash.getSlot(presenceChannel);
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java
index 887f8ee33..3fc85448a 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java
@@ -115,10 +115,7 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
             connection.addListener(this);
             connection.getResources().eventBus().get()
                     .filter(event -> event instanceof ClusterTopologyChangedEvent)
-                    .handle((event, sink) -> {
-                        resubscribeAll();
-                        sink.next(event);
-                    });
+                    .subscribe(event -> resubscribeAll());
         });
     }