Merge branch 'master' into 3.0.0

# Conflicts:
#	redisson/src/main/java/org/redisson/spring/session/RedissonSessionRepository.java

redisson/src/main/java/org/redisson/cache/LocalCacheView.java

@@ -65,6 +65,11 @@ public class LocalCacheView<K, V> {
                 public K next() {
                     return (K) iter.next().getKey();
                 }
+                
+                @Override
+                public void remove() {
+                    iter.remove();
+                }
             };
         }
 
@@ -111,6 +116,11 @@ public class LocalCacheView<K, V> {
                 public V next() {
                     return (V) iter.next().getValue();
                 }
+                
+                @Override
+                public void remove() {
+                    iter.remove();
+                }
             };
         }
 
@@ -152,6 +162,11 @@ public class LocalCacheView<K, V> {
                     CacheValue e = iter.next();
                     return new AbstractMap.SimpleEntry<K, V>((K)e.getKey(), (V)e.getValue());
                 }
+                
+                @Override
+                public void remove() {
+                    iter.remove();
+                }
             };
         }
 

redisson/src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java

@@ -249,7 +249,12 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
         }
         RedisConnection connection = nodeConnections.get(key);
         if (connection != null) {
-            return RedissonPromise.newSucceededFuture(connection);
+            if (!connection.isActive()) {
+                nodeConnections.remove(key);
+                connection.closeAsync();
+            } else {
+                return RedissonPromise.newSucceededFuture(connection);
+            }
         }
 
         if (addr != null) {

redisson/src/main/java/org/redisson/connection/ReplicatedConnectionManager.java

@@ -32,6 +32,7 @@ import org.redisson.config.Config;
 import org.redisson.config.MasterSlaveServersConfig;
 import org.redisson.config.ReadMode;
 import org.redisson.config.ReplicatedServersConfig;
+import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -119,15 +120,15 @@ public class ReplicatedConnectionManager extends MasterSlaveConnectionManager {
         monitorFuture = group.schedule(new Runnable() {
             @Override
             public void run() {
+                if (isShuttingDown()) {
+                    return;
+                }
+
                 final URI master = currentMaster.get();
                 log.debug("Current master: {}", master);
                 
                 final AtomicInteger count = new AtomicInteger(cfg.getNodeAddresses().size());
                 for (final URI addr : cfg.getNodeAddresses()) {
-                    if (isShuttingDown()) {
-                        return;
-                    }
-
                     RFuture<RedisConnection> connectionFuture = connectToNode(cfg, addr, null, addr.getHost());
                     connectionFuture.addListener(new FutureListener<RedisConnection>() {
                         @Override
@@ -166,6 +167,8 @@ public class ReplicatedConnectionManager extends MasterSlaveConnectionManager {
                                         } else if (currentMaster.compareAndSet(master, addr)) {
                                             changeMaster(singleSlotRange.getStartSlot(), addr);
                                         }
+                                    } else if (!config.checkSkipSlavesInit()) {
+                                        slaveUp(addr);
                                     }
                                     
                                     if (count.decrementAndGet() == 0) {
@@ -181,6 +184,13 @@ public class ReplicatedConnectionManager extends MasterSlaveConnectionManager {
         }, cfg.getScanInterval(), TimeUnit.MILLISECONDS);
     }
 
+    private void slaveUp(URI uri) {
+        MasterSlaveEntry entry = getEntry(singleSlotRange.getStartSlot());
+        if (entry.slaveUp(uri, FreezeReason.MANAGER)) {
+            log.info("slave: {} has up", uri);
+        }
+    }
+    
     @Override
     public void shutdown() {
         if (monitorFuture != null) {

redisson/src/main/java/org/redisson/spring/session/RedissonSessionRepository.java

@@ -257,22 +257,22 @@ public class RedissonSessionRepository implements FindByIndexNameSessionReposito
                 publishEvent(new SessionCreatedEvent(this, session));
             }
         } else if (deletedTopic.getPatternNames().contains(pattern.toString())) {
-            if (!body.contains(":")) {
+            if (!body.startsWith(keyPrefix)) {
                 return;
             }
             
-            String id = body.split(":")[1];
+            String id = body.split(keyPrefix)[1];
             RedissonSession session = new RedissonSession(id);
             if (session.load()) {
                 session.clearPrincipal();
             }
             publishEvent(new SessionDeletedEvent(this, session));
         } else if (expiredTopic.getPatternNames().contains(pattern.toString())) {
-            if (!body.contains(":")) {
+            if (!body.startsWith(keyPrefix)) {
                 return;
             }
 
-            String id = body.split(":")[1];
+            String id = body.split(keyPrefix)[1];
             RedissonSession session = new RedissonSession(id);
             if (session.load()) {
                 session.clearPrincipal();