Fixed - Config can not handle underscore in host #722

redisson/src/main/java/org/redisson/client/RedisClient.java

@@ -16,7 +16,7 @@
 package org.redisson.client;
 
 import java.net.InetSocketAddress;
-import java.net.URI;
+import java.net.URL;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -31,7 +31,7 @@
 import org.redisson.client.protocol.RedisCommands;
 import org.redisson.misc.RPromise;
 import org.redisson.misc.RedissonPromise;
-import org.redisson.misc.URIBuilder;
+import org.redisson.misc.URLBuilder;
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
@@ -70,15 +70,15 @@ public class RedisClient {
     private boolean hasOwnGroup;
 
     public RedisClient(String address) {
-        this(URIBuilder.create(address));
+        this(URLBuilder.create(address));
     }
 
-    public RedisClient(URI address) {
+    public RedisClient(URL address) {
         this(new HashedWheelTimer(), Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2), new NioEventLoopGroup(), address);
         hasOwnGroup = true;
     }
 
-    public RedisClient(Timer timer, ExecutorService executor, EventLoopGroup group, URI address) {
+    public RedisClient(Timer timer, ExecutorService executor, EventLoopGroup group, URL address) {
         this(timer, executor, group, address.getHost(), address.getPort());
     }
 

redisson/src/main/java/org/redisson/cluster/ClusterConnectionManager.java

@@ -15,7 +15,7 @@
  */
 package org.redisson.cluster;
 
-import java.net.URI;
+import java.net.URL;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
@@ -66,13 +66,13 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
 
     private final Logger log = LoggerFactory.getLogger(getClass());
 
-    private final Map<URI, RedisConnection> nodeConnections = PlatformDependent.newConcurrentHashMap();
+    private final Map<URL, RedisConnection> nodeConnections = PlatformDependent.newConcurrentHashMap();
 
     private final ConcurrentMap<Integer, ClusterPartition> lastPartitions = PlatformDependent.newConcurrentHashMap();
 
     private ScheduledFuture<?> monitorFuture;
 
-    private volatile URI lastClusterNode;
+    private volatile URL lastClusterNode;
 
     public ClusterConnectionManager(ClusterServersConfig cfg, Config config) {
         super(config);
@@ -84,7 +84,7 @@ public ClusterConnectionManager(ClusterServersConfig cfg, Config config) {
 
         Throwable lastException = null;
         List<String> failedMasters = new ArrayList<String>();
-        for (URI addr : cfg.getNodeAddresses()) {
+        for (URL addr : cfg.getNodeAddresses()) {
             RFuture<RedisConnection> connectionFuture = connect(cfg, addr);
             try {
                 RedisConnection connection = connectionFuture.syncUninterruptibly().getNow();
@@ -158,7 +158,7 @@ private void close(RedisConnection conn) {
         }
     }
 
-    private RFuture<RedisConnection> connect(ClusterServersConfig cfg, final URI addr) {
+    private RFuture<RedisConnection> connect(ClusterServersConfig cfg, final URL addr) {
         RedisConnection connection = nodeConnections.get(addr);
         if (connection != null) {
             return newSucceededFuture(connection);
@@ -308,22 +308,22 @@ public void operationComplete(Future<Void> future) throws Exception {
         return result;
     }
 
-    private void scheduleClusterChangeCheck(final ClusterServersConfig cfg, final Iterator<URI> iterator) {
+    private void scheduleClusterChangeCheck(final ClusterServersConfig cfg, final Iterator<URL> iterator) {
         monitorFuture = GlobalEventExecutor.INSTANCE.schedule(new Runnable() {
             @Override
             public void run() {
                 AtomicReference<Throwable> lastException = new AtomicReference<Throwable>();
-                Iterator<URI> nodesIterator = iterator;
+                Iterator<URL> nodesIterator = iterator;
                 if (nodesIterator == null) {
-                    List<URI> nodes = new ArrayList<URI>();
-                    List<URI> slaves = new ArrayList<URI>();
+                    List<URL> nodes = new ArrayList<URL>();
+                    List<URL> slaves = new ArrayList<URL>();
 
                     for (ClusterPartition partition : getLastPartitions()) {
                         if (!partition.isMasterFail()) {
                             nodes.add(partition.getMasterAddress());
                         }
 
-                        Set<URI> partitionSlaves = new HashSet<URI>(partition.getSlaveAddresses());
+                        Set<URL> partitionSlaves = new HashSet<URL>(partition.getSlaveAddresses());
                         partitionSlaves.removeAll(partition.getFailedSlaveAddresses());
                         slaves.addAll(partitionSlaves);
                     }
@@ -339,7 +339,7 @@ public void run() {
         }, cfg.getScanInterval(), TimeUnit.MILLISECONDS);
     }
 
-    private void checkClusterState(final ClusterServersConfig cfg, final Iterator<URI> iterator, final AtomicReference<Throwable> lastException) {
+    private void checkClusterState(final ClusterServersConfig cfg, final Iterator<URL> iterator, final AtomicReference<Throwable> lastException) {
         if (!iterator.hasNext()) {
             log.error("Can't update cluster state", lastException.get());
             scheduleClusterChangeCheck(cfg, null);
@@ -348,7 +348,7 @@ private void checkClusterState(final ClusterServersConfig cfg, final Iterator<UR
         if (!getShutdownLatch().acquire()) {
             return;
         }
-        final URI uri = iterator.next();
+        final URL uri = iterator.next();
         RFuture<RedisConnection> connectionFuture = connect(cfg, uri);
         connectionFuture.addListener(new FutureListener<RedisConnection>() {
             @Override
@@ -366,7 +366,7 @@ public void operationComplete(Future<RedisConnection> future) throws Exception {
         });
     }
 
-    private void updateClusterState(final ClusterServersConfig cfg, final RedisConnection connection, final Iterator<URI> iterator, final URI uri) {
+    private void updateClusterState(final ClusterServersConfig cfg, final RedisConnection connection, final Iterator<URL> iterator, final URL uri) {
         RFuture<List<ClusterNodeInfo>> future = connection.async(RedisCommands.CLUSTER_NODES);
         future.addListener(new FutureListener<List<ClusterNodeInfo>>() {
             @Override
@@ -415,7 +415,7 @@ private void checkSlaveNodesChange(Collection<ClusterPartition> newPartitions) {
 
                 MasterSlaveEntry entry = getEntry(currentPart.getMasterAddr());
                 // should be invoked first in order to remove stale failedSlaveAddresses
-                Set<URI> addedSlaves = addRemoveSlaves(entry, currentPart, newPart);
+                Set<URL> addedSlaves = addRemoveSlaves(entry, currentPart, newPart);
                 // Do some slaves have changed state from failed to alive?
                 upDownSlaves(entry, currentPart, newPart, addedSlaves);
 
@@ -424,42 +424,42 @@ private void checkSlaveNodesChange(Collection<ClusterPartition> newPartitions) {
         }
     }
 
-    private void upDownSlaves(final MasterSlaveEntry entry, final ClusterPartition currentPart, final ClusterPartition newPart, Set<URI> addedSlaves) {
-        Set<URI> aliveSlaves = new HashSet<URI>(currentPart.getFailedSlaveAddresses());
+    private void upDownSlaves(final MasterSlaveEntry entry, final ClusterPartition currentPart, final ClusterPartition newPart, Set<URL> addedSlaves) {
+        Set<URL> aliveSlaves = new HashSet<URL>(currentPart.getFailedSlaveAddresses());
         aliveSlaves.removeAll(addedSlaves);
         aliveSlaves.removeAll(newPart.getFailedSlaveAddresses());
-        for (URI uri : aliveSlaves) {
+        for (URL uri : aliveSlaves) {
             currentPart.removeFailedSlaveAddress(uri);
             if (entry.slaveUp(uri.getHost(), uri.getPort(), FreezeReason.MANAGER)) {
                 log.info("slave: {} has up for slot ranges: {}", uri, currentPart.getSlotRanges());
             }
         }
 
-        Set<URI> failedSlaves = new HashSet<URI>(newPart.getFailedSlaveAddresses());
+        Set<URL> failedSlaves = new HashSet<URL>(newPart.getFailedSlaveAddresses());
         failedSlaves.removeAll(currentPart.getFailedSlaveAddresses());
-        for (URI uri : failedSlaves) {
+        for (URL uri : failedSlaves) {
             currentPart.addFailedSlaveAddress(uri);
             if (entry.slaveDown(uri.getHost(), uri.getPort(), FreezeReason.MANAGER)) {
                 log.warn("slave: {} has down for slot ranges: {}", uri, currentPart.getSlotRanges());
             }
         }
     }
 
-    private Set<URI> addRemoveSlaves(final MasterSlaveEntry entry, final ClusterPartition currentPart, final ClusterPartition newPart) {
-        Set<URI> removedSlaves = new HashSet<URI>(currentPart.getSlaveAddresses());
+    private Set<URL> addRemoveSlaves(final MasterSlaveEntry entry, final ClusterPartition currentPart, final ClusterPartition newPart) {
+        Set<URL> removedSlaves = new HashSet<URL>(currentPart.getSlaveAddresses());
         removedSlaves.removeAll(newPart.getSlaveAddresses());
 
-        for (URI uri : removedSlaves) {
+        for (URL uri : removedSlaves) {
             currentPart.removeSlaveAddress(uri);
 
             if (entry.slaveDown(uri.getHost(), uri.getPort(), FreezeReason.MANAGER)) {
                 log.info("slave {} removed for slot ranges: {}", uri, currentPart.getSlotRanges());
             }
         }
 
-        Set<URI> addedSlaves = new HashSet<URI>(newPart.getSlaveAddresses());
+        Set<URL> addedSlaves = new HashSet<URL>(newPart.getSlaveAddresses());
         addedSlaves.removeAll(currentPart.getSlaveAddresses());
-        for (final URI uri : addedSlaves) {
+        for (final URL uri : addedSlaves) {
             RFuture<Void> future = entry.addSlave(uri.getHost(), uri.getPort());
             future.addListener(new FutureListener<Void>() {
                 @Override
@@ -516,8 +516,8 @@ private RFuture<Void> checkMasterNodesChange(ClusterServersConfig cfg, Collectio
                     if (!newMasterPart.getMasterAddress().equals(currentPart.getMasterAddress())) {
                         log.info("changing master from {} to {} for {}",
                                 currentPart.getMasterAddress(), newMasterPart.getMasterAddress(), slot);
-                        URI newUri = newMasterPart.getMasterAddress();
-                        URI oldUri = currentPart.getMasterAddress();
+                        URL newUri = newMasterPart.getMasterAddress();
+                        URL oldUri = currentPart.getMasterAddress();
 
                         changeMaster(slot, newUri.getHost(), newUri.getPort());
 
@@ -720,7 +720,7 @@ private HashSet<ClusterPartition> getLastPartitions() {
     }
 
     @Override
-    public URI getLastClusterNode() {
+    public URL getLastClusterNode() {
         return lastClusterNode;
     }
 

redisson/src/main/java/org/redisson/cluster/ClusterNodeInfo.java

@@ -16,10 +16,11 @@
 package org.redisson.cluster;
 
 import java.net.URI;
+import java.net.URL;
 import java.util.HashSet;
 import java.util.Set;
 
-import org.redisson.misc.URIBuilder;
+import org.redisson.misc.URLBuilder;
 
 /**
  * 
@@ -33,7 +34,7 @@ public enum Flag {NOFLAGS, SLAVE, MASTER, MYSELF, FAIL, HANDSHAKE, NOADDR};
     private final String nodeInfo;
 
     private String nodeId;
-    private URI address;
+    private URL address;
     private final Set<Flag> flags = new HashSet<Flag>();
     private String slaveOf;
 
@@ -50,11 +51,11 @@ public void setNodeId(String nodeId) {
         this.nodeId = nodeId;
     }
 
-    public URI getAddress() {
+    public URL getAddress() {
         return address;
     }
     public void setAddress(String address) {
-        this.address = URIBuilder.create(address);
+        this.address = URLBuilder.create(address);
     }
 
     public void addSlotRange(ClusterSlotRange range) {

redisson/src/main/java/org/redisson/cluster/ClusterPartition.java

@@ -16,20 +16,25 @@
 package org.redisson.cluster;
 
 import java.net.InetSocketAddress;
-import java.net.URI;
+import java.net.URL;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
 
-import org.redisson.misc.URIBuilder;
+import org.redisson.misc.URLBuilder;
 
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
 public class ClusterPartition {
 
     private final String nodeId;
     private boolean masterFail;
-    private URI masterAddress;
-    private final Set<URI> slaveAddresses = new HashSet<URI>();
-    private final Set<URI> failedSlaves = new HashSet<URI>();
+    private URL masterAddress;
+    private final Set<URL> slaveAddresses = new HashSet<URL>();
+    private final Set<URL> failedSlaves = new HashSet<URL>();
 
     private final Set<Integer> slots = new HashSet<Integer>();
     private final Set<ClusterSlotRange> slotRanges = new HashSet<ClusterSlotRange>();
@@ -85,33 +90,33 @@ public InetSocketAddress getMasterAddr() {
         return new InetSocketAddress(masterAddress.getHost(), masterAddress.getPort());
     }
 
-    public URI getMasterAddress() {
+    public URL getMasterAddress() {
         return masterAddress;
     }
     public void setMasterAddress(String masterAddress) {
-        setMasterAddress(URIBuilder.create(masterAddress));
+        setMasterAddress(URLBuilder.create(masterAddress));
     }
-    public void setMasterAddress(URI masterAddress) {
+    public void setMasterAddress(URL masterAddress) {
         this.masterAddress = masterAddress;
     }
 
-    public void addFailedSlaveAddress(URI address) {
+    public void addFailedSlaveAddress(URL address) {
         failedSlaves.add(address);
     }
-    public Set<URI> getFailedSlaveAddresses() {
+    public Set<URL> getFailedSlaveAddresses() {
         return Collections.unmodifiableSet(failedSlaves);
     }
-    public void removeFailedSlaveAddress(URI uri) {
+    public void removeFailedSlaveAddress(URL uri) {
         failedSlaves.remove(uri);
     }
 
-    public void addSlaveAddress(URI address) {
+    public void addSlaveAddress(URL address) {
         slaveAddresses.add(address);
     }
-    public Set<URI> getSlaveAddresses() {
+    public Set<URL> getSlaveAddresses() {
         return Collections.unmodifiableSet(slaveAddresses);
     }
-    public void removeSlaveAddress(URI uri) {
+    public void removeSlaveAddress(URL uri) {
         slaveAddresses.remove(uri);
         failedSlaves.remove(uri);
     }

redisson/src/main/java/org/redisson/config/ClusterServersConfig.java

@@ -15,18 +15,23 @@
  */
 package org.redisson.config;
 
-import java.net.URI;
+import java.net.URL;
 import java.util.ArrayList;
 import java.util.List;
 
-import org.redisson.misc.URIBuilder;
+import org.redisson.misc.URLBuilder;
 
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
 public class ClusterServersConfig extends BaseMasterSlaveServersConfig<ClusterServersConfig> {
 
     /**
      * Redis cluster node urls list
      */
-    private List<URI> nodeAddresses = new ArrayList<URI>();
+    private List<URL> nodeAddresses = new ArrayList<URL>();
 
     /**
      * Redis cluster scan interval in milliseconds
@@ -50,14 +55,14 @@ public ClusterServersConfig() {
      */
     public ClusterServersConfig addNodeAddress(String ... addresses) {
         for (String address : addresses) {
-            nodeAddresses.add(URIBuilder.create(address));
+            nodeAddresses.add(URLBuilder.create(address));
         }
         return this;
     }
-    public List<URI> getNodeAddresses() {
+    public List<URL> getNodeAddresses() {
         return nodeAddresses;
     }
-    void setNodeAddresses(List<URI> nodeAddresses) {
+    void setNodeAddresses(List<URL> nodeAddresses) {
         this.nodeAddresses = nodeAddresses;
     }
 

redisson/src/main/java/org/redisson/config/ConfigSupport.java

@@ -47,7 +47,13 @@
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
 import org.redisson.codec.CodecProvider;
 import org.redisson.liveobject.provider.ResolverProvider;
+import org.redisson.misc.URLBuilder;
 
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
 public class ConfigSupport {
 
     @JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, property = "class")
@@ -111,6 +117,10 @@ public static class ConfigMixIn {
     private final ObjectMapper jsonMapper = createMapper(null);
     private final ObjectMapper yamlMapper = createMapper(new YAMLFactory());
 
+    public ConfigSupport() {
+        URLBuilder.init();
+    }
+
     public <T> T fromJSON(String content, Class<T> configType) throws IOException {
         return jsonMapper.readValue(content, configType);
     }