Connections restore with Sentinel servers fixed.

redisson · Dec 25, 2015 · 76eebdd · 76eebdd
1 parent c20e550
commit 76eebdd
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 28 deletions.
diff --git a/src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java b/src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java
@@ -545,7 +545,6 @@ public void operationComplete(Future<PubSubConnectionEntry> future)
     protected void slaveDown(ClusterSlotRange slotRange, String host, int port, FreezeReason freezeReason) {
         MasterSlaveEntry entry = getEntry(slotRange);
         slaveDown(entry, host, port, freezeReason);
-        log.info("slave: {}:{} has down", host, port);
     }
 
     protected void changeMaster(ClusterSlotRange slotRange, String host, int port) {

diff --git a/src/main/java/org/redisson/connection/MasterSlaveEntry.java b/src/main/java/org/redisson/connection/MasterSlaveEntry.java
@@ -92,8 +92,9 @@ public Collection<RedisPubSubConnection> slaveDown(String host, int port, Freeze
         // add master as slave if no more slaves available
         if (slaveBalancer.getAvailableClients() == 0) {
             InetSocketAddress addr = masterEntry.getClient().getAddr();
-            slaveUp(addr.getHostName(), addr.getPort(), FreezeReason.SYSTEM);
+            if (slaveUp(addr.getHostName(), addr.getPort(), FreezeReason.SYSTEM)) {
-            log.info("master {}:{} used as slave", addr.getHostName(), addr.getPort());
+                log.info("master {}:{} used as slave", addr.getHostName(), addr.getPort());
+            }
         }
         return conns;
     }
@@ -120,15 +121,17 @@ public RedisClient getClient() {
         return masterEntry.getClient();
     }
 
-    public void slaveUp(String host, int port, FreezeReason freezeReason) {
+    public boolean slaveUp(String host, int port, FreezeReason freezeReason) {
         if (!slaveBalancer.unfreeze(host, port, freezeReason)) {
-            return;
+            return false;
         }
+
         InetSocketAddress addr = masterEntry.getClient().getAddr();
         // exclude master from slaves
         if (!addr.getHostName().equals(host) || port != addr.getPort()) {
             connectionManager.slaveDown(this, addr.getHostName(), addr.getPort(), FreezeReason.SYSTEM);
         }
+        return true;
     }
 
     /**

diff --git a/src/main/java/org/redisson/connection/SentinelConnectionManager.java b/src/main/java/org/redisson/connection/SentinelConnectionManager.java
@@ -49,7 +49,6 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
 
     private final ConcurrentMap<String, RedisClient> sentinels = PlatformDependent.newConcurrentHashMap();
     private final AtomicReference<String> currentMaster = new AtomicReference<String>();
-    private final ConcurrentMap<String, Boolean> freezeSlaves = PlatformDependent.newConcurrentHashMap();
     private final ConcurrentMap<String, Boolean> slaves = PlatformDependent.newConcurrentHashMap();
 
 
@@ -222,8 +221,10 @@ protected void onSlaveAdded(URI addr, String msg) {
 
             // to avoid addition twice
             if (slaves.putIfAbsent(slaveAddr, true) == null) {
-                addSlave(ip, Integer.valueOf(port));
+                getEntry(singleSlotRange).addSlave(ip, Integer.valueOf(port));
                 log.info("slave: {} added", slaveAddr);
+            } else {
+                slaveUp(ip, port);
             }
         } else {
             log.warn("onSlaveAdded. Invalid message: {} from Sentinel {}:{}", msg, addr.getHost(), addr.getPort());
@@ -266,11 +267,8 @@ private void onNodeDown(URI sentinelAddr, String msg) {
     }
 
     private void slaveDown(String ip, String port) {
-        // to avoid freeze twice
+        slaveDown(singleSlotRange, ip, Integer.valueOf(port), FreezeReason.MANAGER);
-        String addr = ip + ":" + port;
+        log.info("slave: {}:{} has down", ip, port);
-        if (freezeSlaves.putIfAbsent(addr, true) == null) {
-            slaveDown(singleSlotRange, ip, Integer.valueOf(port), FreezeReason.MANAGER);
-        }
     }
 
     private void onNodeUp(URI addr, String msg) {
@@ -281,11 +279,7 @@ private void onNodeUp(URI addr, String msg) {
                 String ip = parts[2];
                 String port = parts[3];
 
-                String slaveAddr = ip + ":" + port;
+                slaveUp(ip, port);
-                if (freezeSlaves.remove(slaveAddr) != null) {
-                    slaveUp(ip, Integer.valueOf(port));
-                    log.info("slave: {} has up", slaveAddr);
-                }
             } else if ("master".equals(parts[0])) {
                 String ip = parts[2];
                 String port = parts[3];
@@ -303,6 +297,13 @@ private void onNodeUp(URI addr, String msg) {
         }
     }
 
+    private void slaveUp(String ip, String port) {
+        if (getEntry(singleSlotRange).slaveUp(ip, Integer.valueOf(port), FreezeReason.MANAGER)) {
+            String slaveAddr = ip + ":" + port;
+            log.info("slave: {} has up", slaveAddr);
+        }
+    }
+
     private void onMasterChange(SentinelServersConfig cfg, URI addr, String msg) {
         String[] parts = msg.split(" ");
 
@@ -324,14 +325,6 @@ private void onMasterChange(SentinelServersConfig cfg, URI addr, String msg) {
         }
     }
 
-    private void addSlave(String host, int port) {
-        getEntry(0).addSlave(host, port);
-    }
-
-    private void slaveUp(String host, int port) {
-        getEntry(0).slaveUp(host, port, FreezeReason.MANAGER);
-    }
-
     @Override
     public void shutdown() {
         super.shutdown();

diff --git a/src/main/java/org/redisson/connection/balancer/LoadBalancerManagerImpl.java b/src/main/java/org/redisson/connection/balancer/LoadBalancerManagerImpl.java
@@ -134,9 +134,11 @@ public Collection<RedisPubSubConnection> freeze(String host, int port, FreezeRea
             connection.closeAsync();
         }
 
-        List<RedisPubSubConnection> list = new ArrayList<RedisPubSubConnection>(connectionEntry.getAllSubscribeConnections());
+        synchronized (connectionEntry) {
-        connectionEntry.getAllSubscribeConnections().clear();
+            List<RedisPubSubConnection> list = new ArrayList<RedisPubSubConnection>(connectionEntry.getAllSubscribeConnections());
-        return list;
+            connectionEntry.getAllSubscribeConnections().clear();
+            return list;
+        }
     }
 
     public Future<RedisPubSubConnection> nextPubSubConnection() {