Make RedisClusterMessagesCache a Managed class.

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -349,6 +349,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     environment.lifecycle().manage(messagePersister);
     environment.lifecycle().manage(accountDatabaseCrawler);
     environment.lifecycle().manage(remoteConfigsManager);
+    environment.lifecycle().manage(clusterMessagesCache);
     environment.lifecycle().manage(clusterMessagePersister);
     environment.lifecycle().manage(clientPresenceManager);
 

service/src/main/java/org/whispersystems/textsecuregcm/storage/RedisClusterMessagesCache.java

@@ -2,6 +2,7 @@ package org.whispersystems.textsecuregcm.storage;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.InvalidProtocolBufferException;
+import io.dropwizard.lifecycle.Managed;
 import io.lettuce.core.ScriptOutputType;
 import io.lettuce.core.cluster.SlotHash;
 import io.lettuce.core.cluster.event.ClusterTopologyChangedEvent;
@@ -36,7 +37,7 @@ import java.util.concurrent.ExecutorService;
 
 import static com.codahale.metrics.MetricRegistry.name;
 
-public class RedisClusterMessagesCache extends RedisClusterPubSubAdapter<String, String> implements UserMessagesCache {
+public class RedisClusterMessagesCache extends RedisClusterPubSubAdapter<String, String> implements UserMessagesCache, Managed {
 
     private final FaultTolerantRedisCluster                     redisCluster;
     private final FaultTolerantPubSubConnection<String, String> pubSubConnection;
@@ -90,7 +91,10 @@ public class RedisClusterMessagesCache extends RedisClusterPubSubAdapter<String,
         this.getItemsScript           = ClusterLuaScript.fromResource(redisCluster, "lua/get_items.lua",             ScriptOutputType.MULTI);
         this.removeQueueScript        = ClusterLuaScript.fromResource(redisCluster, "lua/remove_queue.lua",          ScriptOutputType.STATUS);
         this.getQueuesToPersistScript = ClusterLuaScript.fromResource(redisCluster, "lua/get_queues_to_persist.lua", ScriptOutputType.MULTI);
+    }
 
+    @Override
+    public void start() {
         pubSubConnection.usePubSubConnection(connection -> {
             connection.addListener(this);
             connection.getResources().eventBus().get()
@@ -102,6 +106,11 @@ public class RedisClusterMessagesCache extends RedisClusterPubSubAdapter<String,
         });
     }
 
+    @Override
+    public void stop() {
+        pubSubConnection.usePubSubConnection(connection -> connection.sync().masters().commands().unsubscribe());
+    }
+
     private void resubscribeAll() {
         final Set<String> queueNames;
 

service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisClusterMessagesCacheTest.java

@@ -35,14 +35,18 @@ public class RedisClusterMessagesCacheTest extends AbstractMessagesCacheTest {
 
         notificationExecutorService = Executors.newSingleThreadExecutor();
         messagesCache               = new RedisClusterMessagesCache(getRedisCluster(), notificationExecutorService);
+
+        messagesCache.start();
     }
 
     @Override
     public void tearDown() throws Exception {
-        super.tearDown();
+        messagesCache.stop();
 
         notificationExecutorService.shutdown();
         notificationExecutorService.awaitTermination(1, TimeUnit.SECONDS);
+
+        super.tearDown();
     }
 
     @Override