Merge branch 'master' into 3.0.0

README.md

@@ -148,7 +148,7 @@ RExecutorService executor = redisson.getExecutorService("myExecutorService");
 
 ```
 
-Please consider __[Redisson PRO](https://redisson.pro)__ version for advanced features and support by SLA.
+<sub>Please consider __[Redisson PRO](https://redisson.pro)__ version for advanced features and support by SLA.</sub>
 
 Downloads
 ===============================

redisson-spring-boot-starter/README.md

@@ -5,6 +5,8 @@ Integrates Redisson with Spring Boot library. Depends on [Spring Data Redis](htt
 
 Supports Spring Boot 1.3.x, 1.4.x, 1.5.x, 2.0.x
 
+<sub>Please consider __[Redisson PRO](https://redisson.pro)__ version for advanced features and support by SLA.</sub>
+
 Usage
 ===
 
@@ -64,3 +66,4 @@ spring.redis.redisson.config=classpath:redisson.yaml
 ```
 
 ### 3. Get access to Redisson through spring bean with `RedissonClient` interface
+

redisson-spring-data/README.md

@@ -5,6 +5,8 @@ Integrates Redisson with Spring Data Redis library. Provides ability to work wit
 
 Supports Spring Data Redis 1.6.x, 1.7.x, 1.8.x, 2.0.x
 
+<sub>Please consider __[Redisson PRO](https://redisson.pro)__ version for advanced features and support by SLA.</sub>
+
 Usage
 ===
 

redisson-tomcat/README.md

@@ -5,6 +5,8 @@ Stores session of Apache Tomcat in Redis and allows to distribute requests acros
 
 Supports Apache Tomcat 6.x, 7.x, 8.x, 9.x
 
+<sub>Please consider __[Redisson PRO](https://redisson.pro)__ version for advanced features and support by SLA.</sub>
+
 Advantages
 ===
 

redisson/src/main/java/org/redisson/client/protocol/CommandData.java

@@ -97,9 +97,8 @@ public class CommandData<T, R> implements QueueCommand {
     }
     
     public boolean isBlockingCommand() {
-        return RedisCommands.BLOCKING_COMMANDS.contains(command.getName()) 
+        return RedisCommands.BLOCKING_COMMAND_NAMES.contains(command.getName()) 
-                || RedisCommands.XREAD_BLOCKING_SINGLE == command
+                || RedisCommands.BLOCKING_COMMANDS.contains(command);
-                    || RedisCommands.XREAD_BLOCKING == command;
     }
 
     @Override

redisson/src/main/java/org/redisson/client/protocol/RedisCommands.java

@@ -24,6 +24,7 @@ import java.util.Set;
 
 import org.redisson.api.RType;
 import org.redisson.api.StreamId;
+import org.redisson.client.codec.StringCodec;
 import org.redisson.client.protocol.RedisCommand.ValueType;
 import org.redisson.client.protocol.convertor.BitsSizeReplayConvertor;
 import org.redisson.client.protocol.convertor.BooleanAmountReplayConvertor;
@@ -53,8 +54,10 @@ import org.redisson.client.protocol.decoder.Long2MultiDecoder;
 import org.redisson.client.protocol.decoder.LongMultiDecoder;
 import org.redisson.client.protocol.decoder.MapScanResult;
 import org.redisson.client.protocol.decoder.MapScanResultReplayDecoder;
+import org.redisson.client.protocol.decoder.ObjectDecoder;
 import org.redisson.client.protocol.decoder.ObjectFirstScoreReplayDecoder;
 import org.redisson.client.protocol.decoder.ObjectListReplayDecoder;
+import org.redisson.client.protocol.decoder.ObjectMapDecoder;
 import org.redisson.client.protocol.decoder.ObjectMapEntryReplayDecoder;
 import org.redisson.client.protocol.decoder.ObjectMapJoinDecoder;
 import org.redisson.client.protocol.decoder.ObjectMapReplayDecoder;
@@ -66,6 +69,7 @@ import org.redisson.client.protocol.decoder.ScoredSortedSetReplayDecoder;
 import org.redisson.client.protocol.decoder.ScoredSortedSetScanDecoder;
 import org.redisson.client.protocol.decoder.ScoredSortedSetScanReplayDecoder;
 import org.redisson.client.protocol.decoder.SlotsDecoder;
+import org.redisson.client.protocol.decoder.StreamIdDecoder;
 import org.redisson.client.protocol.decoder.StreamResultDecoder;
 import org.redisson.client.protocol.decoder.StringDataDecoder;
 import org.redisson.client.protocol.decoder.StringListReplayDecoder;
@@ -209,7 +213,7 @@ public interface RedisCommands {
     RedisCommand<Object> BZPOPMIN_VALUE = new RedisCommand<Object>("BZPOPMIN", new ScoredSortedSetPolledObjectDecoder());
     RedisCommand<Object> BZPOPMAX_VALUE = new RedisCommand<Object>("BZPOPMAX", new ScoredSortedSetPolledObjectDecoder());
 
-    Set<String> BLOCKING_COMMANDS = new HashSet<String>(
+    Set<String> BLOCKING_COMMAND_NAMES = new HashSet<String>(
             Arrays.asList(BLPOP_VALUE.getName(), BRPOP_VALUE.getName(), BRPOPLPUSH.getName(), BZPOPMIN_VALUE.getName(), BZPOPMAX_VALUE.getName(), 
                     BLPOP.getName(), BRPOP.getName()));
 
@@ -321,50 +325,82 @@ public interface RedisCommands {
     RedisCommand<Boolean> SETNX = new RedisCommand<Boolean>("SETNX", new BooleanReplayConvertor());
     RedisCommand<Void> PSETEX = new RedisCommand<Void>("PSETEX", new VoidReplayConvertor());
 
-    RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>> XREVRANGE = new RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>>("XREVRANGE",
-            new ListMultiDecoder(new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET, new StreamIdConvertor()), 
-                    new ObjectMapJoinDecoder()));
-    
     RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>> XRANGE = new RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>>("XRANGE",
-            new ListMultiDecoder(new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET, new StreamIdConvertor()), 
+            new ListMultiDecoder(
+                    new ObjectDecoder(new StreamIdDecoder()),
+                    new ObjectMapReplayDecoder(),
+                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET),
                     new ObjectMapJoinDecoder()));
+
+    RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>> XREVRANGE = new RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>>("XREVRANGE",
+            XRANGE.getReplayMultiDecoder());
     
     RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>> XREAD = new RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>>("XREAD",
-            new ListMultiDecoder(new ListResultReplayDecoder(), new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1, new StreamIdConvertor()), 
+            new ListMultiDecoder(
-                    new ObjectMapJoinDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX)));
+                    new ObjectDecoder(StringCodec.INSTANCE.getValueDecoder()),
-    
+                    new ObjectDecoder(new StreamIdDecoder()),
-    RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>> XREAD_BLOCKING = new RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>>("XREAD",
+                    new ObjectMapReplayDecoder(),
-            new ListMultiDecoder(new ListResultReplayDecoder(), new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1, new StreamIdConvertor()), 
+                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
-                    new ObjectMapJoinDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX)));
+                    new ObjectMapJoinDecoder(), 
+                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX),
+                    new StreamResultDecoder()));
+            
+    RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>> XREAD_BLOCKING = new RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>>("XREAD", XREAD.getReplayMultiDecoder());
 
     RedisCommand<Map<StreamId, Map<Object, Object>>> XREAD_SINGLE = new RedisCommand<Map<StreamId, Map<Object, Object>>>("XREAD",
-            new ListMultiDecoder(new ListResultReplayDecoder(), new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1, new StreamIdConvertor()), 
+            new ListMultiDecoder(
-                    new ObjectMapJoinDecoder(), new ObjectMapReplayDecoder(), new StreamResultDecoder()));
+                    new ObjectDecoder(StringCodec.INSTANCE.getValueDecoder()),
+                    new ObjectDecoder(new StreamIdDecoder()),
+                    new ObjectMapReplayDecoder(),
+                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
+                    new ObjectMapJoinDecoder(), 
+                    new ObjectMapReplayDecoder(),
+                    new StreamResultDecoder()));
     
-    RedisCommand<Map<StreamId, Map<Object, Object>>> XREAD_BLOCKING_SINGLE = new RedisCommand<Map<StreamId, Map<Object, Object>>>("XREAD",
+    RedisCommand<Map<StreamId, Map<Object, Object>>> XREAD_BLOCKING_SINGLE = new RedisCommand<Map<StreamId, Map<Object, Object>>>("XREAD", XREAD_SINGLE.getReplayMultiDecoder());
-            new ListMultiDecoder(new ListResultReplayDecoder(), new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1, new StreamIdConvertor()), 
-                    new ObjectMapJoinDecoder(), new ObjectMapReplayDecoder(), new StreamResultDecoder()));
 
     RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>> XREADGROUP = new RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>>("XREADGROUP",
-            new ListMultiDecoder(new ListResultReplayDecoder(), new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1, new StreamIdConvertor()), 
+            new ListMultiDecoder(
-                    new ObjectMapJoinDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX)));
+                    new ObjectDecoder(StringCodec.INSTANCE.getValueDecoder()),
-    
+                    new ObjectDecoder(new StreamIdDecoder()),
-    RedisCommand<Map<StreamId, Map<Object, Object>>> XREADGROUP_BLOCKING = new RedisCommand<Map<StreamId, Map<Object, Object>>>("XREADGROUP",
+                    new ObjectMapReplayDecoder(),
-            new ListMultiDecoder(new ListResultReplayDecoder(), new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1, new StreamIdConvertor()), 
+                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
-                    new ObjectMapJoinDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX)));
+                    new ObjectMapJoinDecoder(), 
+                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX),
+                    new StreamResultDecoder()));
 
+    RedisCommand<Map<StreamId, Map<Object, Object>>> XREADGROUP_BLOCKING = new RedisCommand<Map<StreamId, Map<Object, Object>>>("XREADGROUP",
+            new ListMultiDecoder(
+                    new ObjectDecoder(StringCodec.INSTANCE.getValueDecoder()),
+                    new ObjectDecoder(new StreamIdDecoder()),
+                    new ObjectMapReplayDecoder(),
+                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
+                    new ObjectMapJoinDecoder(), 
+                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX),
+                    new StreamResultDecoder()));
+            
     RedisCommand<Map<StreamId, Map<Object, Object>>> XREADGROUP_SINGLE = new RedisCommand<Map<StreamId, Map<Object, Object>>>("XREADGROUP",
-            new ListMultiDecoder(new ListResultReplayDecoder(), new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1, new StreamIdConvertor()), 
+            new ListMultiDecoder(
-                    new ObjectMapJoinDecoder(), new ObjectMapReplayDecoder(), new StreamResultDecoder()));
+                    new ObjectDecoder(StringCodec.INSTANCE.getValueDecoder()),
+                    new ObjectDecoder(new StreamIdDecoder()),
+                    new ObjectMapReplayDecoder(),
+                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
+                    new ObjectMapJoinDecoder(), 
+                    new ObjectMapReplayDecoder(),
+                    new StreamResultDecoder()));
 
     RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>> XCLAIM = new RedisCommand<Map<String, Map<StreamId, Map<Object, Object>>>>("XCLAIM",
-            new ListMultiDecoder(new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET, new StreamIdConvertor()), 
+            new ListMultiDecoder(
+                    new ObjectDecoder(new StreamIdDecoder()),
+                    new ObjectMapReplayDecoder(),
+                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET),
                     new ObjectMapJoinDecoder()));
-    
+            
     RedisCommand<Map<StreamId, Map<Object, Object>>> XREADGROUP_BLOCKING_SINGLE = new RedisCommand<Map<StreamId, Map<Object, Object>>>("XREADGROUP",
-            new ListMultiDecoder(new ListResultReplayDecoder(), new ObjectMapReplayDecoder(), new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1, new StreamIdConvertor()), 
+            XREADGROUP_SINGLE.getReplayMultiDecoder());
-                    new ObjectMapJoinDecoder(), new ObjectMapReplayDecoder(), new StreamResultDecoder()));
 
+    Set<RedisCommand> BLOCKING_COMMANDS = new HashSet<RedisCommand>(Arrays.asList(
+            XREAD_BLOCKING_SINGLE, XREAD_BLOCKING, XREADGROUP_BLOCKING_SINGLE, XREADGROUP_BLOCKING));
     
     RedisStrictCommand<StreamId> XADD = new RedisStrictCommand<StreamId>("XADD", new StreamIdConvertor());
     RedisStrictCommand<Void> XGROUP = new RedisStrictCommand<Void>("XGROUP", new VoidReplayConvertor());

redisson/src/main/java/org/redisson/client/protocol/decoder/ObjectDecoder.java

@@ -0,0 +1,47 @@
+/**
+ * Copyright 2018 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.client.protocol.decoder;
+
+import java.util.List;
+
+import org.redisson.client.handler.State;
+import org.redisson.client.protocol.Decoder;
+
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
+public class ObjectDecoder implements MultiDecoder<Object> {
+
+    private Decoder<Object> decoder;
+    
+    public ObjectDecoder(Decoder<Object> decoder) {
+        super();
+        this.decoder = decoder;
+    }
+
+    @Override
+    public Decoder<Object> getDecoder(int paramNum, State state) {
+        return decoder;
+    }
+
+    @Override
+    public Object decode(List<Object> parts, State state) {
+        throw new IllegalStateException();
+    }
+
+}

redisson/src/main/java/org/redisson/client/protocol/decoder/ObjectMapReplayDecoder.java

@@ -21,7 +21,6 @@ import java.util.Map;
 
 import org.redisson.client.handler.State;
 import org.redisson.client.protocol.Decoder;
-import org.redisson.client.protocol.convertor.Convertor;
 
 /**
  * 
@@ -31,7 +30,6 @@ import org.redisson.client.protocol.convertor.Convertor;
 public class ObjectMapReplayDecoder implements MultiDecoder<Map<Object, Object>> {
 
     private Decoder<Object> codec;
-    private Convertor<?> convertor;
     
     public ObjectMapReplayDecoder() {
     }
@@ -41,25 +39,23 @@ public class ObjectMapReplayDecoder implements MultiDecoder<Map<Object, Object>>
         this.codec = codec;
     }
     
-    public ObjectMapReplayDecoder(Decoder<Object> codec, Convertor<?> convertor) {
-        super();
-        this.codec = codec;
-        this.convertor = convertor;
-    }
-
     @Override
     public Map<Object, Object> decode(List<Object> parts, State state) {
-        Map<Object, Object> result = new LinkedHashMap<Object, Object>(parts.size()/2);
+        if (parts.get(0) instanceof Map) {
-        for (int i = 0; i < parts.size(); i++) {
+            Map<Object, Object> result = new LinkedHashMap<Object, Object>(parts.size());
-            if (i % 2 != 0) {
+            for (int i = 0; i < parts.size(); i++) {
-                if (convertor != null) {
+                result.putAll((Map<? extends Object, ? extends Object>) parts.get(i));
-                    result.put(convertor.convert(parts.get(i-1)), parts.get(i));
+            }
-                } else {
+            return result;
+        } else {
+            Map<Object, Object> result = new LinkedHashMap<Object, Object>(parts.size()/2);
+            for (int i = 0; i < parts.size(); i++) {
+                if (i % 2 != 0) {
                     result.put(parts.get(i-1), parts.get(i));
                 }
-           }
+            }
+            return result;
         }
-        return result;
     }
 
     @Override

redisson/src/main/java/org/redisson/client/protocol/decoder/StreamIdDecoder.java

@@ -0,0 +1,41 @@
+/**
+ * Copyright 2018 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.client.protocol.decoder;
+
+import java.io.IOException;
+
+import org.redisson.api.StreamId;
+import org.redisson.client.codec.StringCodec;
+import org.redisson.client.handler.State;
+import org.redisson.client.protocol.Decoder;
+
+import io.netty.buffer.ByteBuf;
+
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
+public class StreamIdDecoder implements Decoder<Object> {
+
+    @Override
+    public Object decode(ByteBuf buf, State state) throws IOException {
+        String id = (String) StringCodec.INSTANCE.getValueDecoder().decode(buf, state);
+        String[] parts = id.toString().split("-");
+        return new StreamId(Long.valueOf(parts[0]), Long.valueOf(parts[1]));
+    }
+
+}

redisson/src/main/java/org/redisson/command/CommandAsyncService.java

@@ -846,12 +846,10 @@ public class CommandAsyncService implements CommandAsyncExecutor {
         details.getTimeout().cancel();
 
         long timeoutTime = connectionManager.getConfig().getTimeout();
-        if (RedisCommands.BLOCKING_COMMANDS.contains(details.getCommand().getName())
+        if (RedisCommands.BLOCKING_COMMAND_NAMES.contains(details.getCommand().getName())
-                || RedisCommands.XREAD_BLOCKING_SINGLE == details.getCommand()
+                || RedisCommands.BLOCKING_COMMANDS.contains(details.getCommand())) {
-                    || RedisCommands.XREAD_BLOCKING == details.getCommand()) {
             Long popTimeout = null;
-            if (RedisCommands.XREAD_BLOCKING_SINGLE == details.getCommand()
+            if (RedisCommands.BLOCKING_COMMANDS.contains(details.getCommand())) {
-                    || RedisCommands.XREAD_BLOCKING == details.getCommand()) {
                 boolean found = false;
                 for (Object param : details.getParams()) {
                     if (found) {

redisson/src/main/java/org/redisson/connection/dns/MultiDnsAddressResolverGroup.java

@@ -19,13 +19,14 @@ import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.List;
 
+import io.netty.channel.ChannelFactory;
+import io.netty.channel.EventLoop;
 import io.netty.channel.socket.DatagramChannel;
 import io.netty.resolver.AddressResolver;
 import io.netty.resolver.dns.DnsAddressResolverGroup;
 import io.netty.resolver.dns.DnsServerAddressStream;
 import io.netty.resolver.dns.DnsServerAddressStreamProvider;
 import io.netty.resolver.dns.SingletonDnsServerAddressStreamProvider;
-import io.netty.util.concurrent.EventExecutor;
 
 /**
  * Workaround for https://github.com/netty/netty/issues/8261
@@ -59,10 +60,12 @@ public class MultiDnsAddressResolverGroup extends DnsAddressResolverGroup {
     }
     
     @Override
-    public AddressResolver<InetSocketAddress> getResolver(EventExecutor executor) {
+    protected AddressResolver<InetSocketAddress> newResolver(EventLoop eventLoop,
+            ChannelFactory<? extends DatagramChannel> channelFactory, DnsServerAddressStreamProvider nameServerProvider)
+            throws Exception {
         List<AddressResolver<InetSocketAddress>> resolvers = new ArrayList<AddressResolver<InetSocketAddress>>();
         for (DnsAddressResolverGroup group : groups) {
-            resolvers.add(group.getResolver(executor));
+            resolvers.add(group.getResolver(eventLoop));
         }
         return new GroupAddressResolver(resolvers);
     }

redisson/src/main/java/org/redisson/misc/URIBuilder.java

@@ -28,6 +28,11 @@ import java.net.URI;
 public class URIBuilder {
 
     public static URI create(String uri) {
+        if (!uri.startsWith("redis://")
+                && !uri.startsWith("rediss://")) {
+            throw new IllegalArgumentException("Redis url should start with redis:// or rediss:// (for SSL connection)");
+        }
+        
         URI u = URI.create(uri);
         // Let's assuming most of the time it is OK.
         if (u.getHost() != null) {