Decoders optimizations

src/main/java/org/redisson/client/handler/RedisDecoder.java

@@ -37,6 +37,12 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ReplayingDecoder;
 import io.netty.util.CharsetUtil;
 
+/**
+ * Code parts from Sam Pullara
+ *
+ * @author Nikita Koksharov
+ *
+ */
 public class RedisDecoder extends ReplayingDecoder<Void> {
 
     public static final char CR = '\r';
@@ -64,7 +70,7 @@ public class RedisDecoder extends ReplayingDecoder<Void> {
 
         try {
             decode(in, data, null, pubSubConnection, currentDecoder);
-        } catch (Exception e) {
+        } catch (IOException e) {
             data.getPromise().setFailure(e);
         }
 

src/main/java/org/redisson/client/handler/RedisEncoder.java

@@ -24,6 +24,12 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToByteEncoder;
 import io.netty.util.CharsetUtil;
 
+/**
+ * Code parts from Sam Pullara
+ *
+ * @author Nikita Koksharov
+ *
+ */
 public class RedisEncoder extends MessageToByteEncoder<RedisData<Object, Object>> {
 
     private final Encoder paramsEncoder = new StringParamsEncoder();
@@ -89,6 +95,9 @@ public class RedisEncoder extends MessageToByteEncoder<RedisData<Object, Object>
         if (msg.getCommand().getInParamType().get(typeIndex) == ValueType.MAP_VALUE) {
             return msg.getCodec().getMapValueEncoder();
         }
+        if (msg.getCommand().getInParamType().get(typeIndex) == ValueType.OBJECTS) {
+            return msg.getCodec().getValueEncoder();
+        }
         throw new IllegalStateException();
     }
 

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

@@ -22,7 +22,7 @@ import org.redisson.client.protocol.decoder.MultiDecoder;
 
 public class RedisCommand<R> {
 
-    public enum ValueType {OBJECT, MAP_VALUE, MAP_KEY, MAP}
+    public enum ValueType {OBJECT, OBJECTS, MAP_VALUE, MAP_KEY, MAP}
 
     private ValueType outParamType = ValueType.OBJECT;
     private List<ValueType> inParamType = Arrays.asList(ValueType.OBJECT);
@@ -96,6 +96,12 @@ public class RedisCommand<R> {
         this.inParamType = Arrays.asList(inParamType);
     }
 
+    public RedisCommand(String name, Convertor<R> convertor, int encodeParamIndex, List<ValueType> inParamTypes) {
+        this(name, null, null, null, encodeParamIndex);
+        this.convertor = convertor;
+        this.inParamType = inParamTypes;
+    }
+
     public RedisCommand(String name, Convertor<R> convertor, int encodeParamIndex) {
         this(name, null, null, null, encodeParamIndex);
         this.convertor = convertor;

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

@@ -22,6 +22,8 @@ import org.redisson.client.protocol.RedisCommand.ValueType;
 import org.redisson.client.protocol.decoder.BooleanStatusReplayDecoder;
 import org.redisson.client.protocol.decoder.KeyValueObjectDecoder;
 import org.redisson.client.protocol.decoder.MapScanResultReplayDecoder;
+import org.redisson.client.protocol.decoder.MapScanResultReplayDecoder2;
+import org.redisson.client.protocol.decoder.NestedMultiDecoder;
 import org.redisson.client.protocol.decoder.ObjectListReplayDecoder;
 import org.redisson.client.protocol.decoder.ObjectMapReplayDecoder;
 import org.redisson.client.protocol.decoder.StringDataDecoder;
@@ -36,6 +38,7 @@ import com.lambdaworks.redis.output.MapScanResult;
 public interface RedisCommands {
 
     RedisCommand<Object> LPOP = new RedisCommand<Object>("LPOP");
+    RedisCommand<Long> LREM = new RedisCommand<Long>("LREM", 3);
     RedisCommand<Object> LINDEX = new RedisCommand<Object>("LINDEX");
     RedisStrictCommand<Long> LLEN = new RedisStrictCommand<Long>("LLEN");
 
@@ -48,10 +51,14 @@ public interface RedisCommands {
     RedisCommand<Object> BRPOPLPUSH = new RedisCommand<Object>("BRPOPLPUSH");
     RedisCommand<Object> BLPOP = new RedisCommand<Object>("BLPOP", new KeyValueObjectDecoder());
 
+    RedisCommand<Boolean> PFADD = new RedisCommand<Boolean>("PFADD", new BooleanReplayConvertor(), 2);
+    RedisCommand<Long> PFCOUNT = new RedisCommand<Long>("PFCOUNT");
+    RedisStrictCommand<String> PFMERGE = new RedisStrictCommand<String>("PFMERGE", new StringReplayDecoder());
+
     RedisCommand<Long> RPOP = new RedisCommand<Long>("RPOP");
     RedisCommand<Long> LPUSH = new RedisCommand<Long>("LPUSH");
     RedisCommand<List<Object>> LRANGE = new RedisCommand<List<Object>>("LRANGE", new ObjectListReplayDecoder());
-    RedisCommand<Long> RPUSH = new RedisCommand<Long>("RPUSH");
+    RedisCommand<Long> RPUSH = new RedisCommand<Long>("RPUSH", 2, ValueType.OBJECTS);
 
     RedisStrictCommand<Boolean> EVAL_BOOLEAN = new RedisStrictCommand<Boolean>("EVAL", new BooleanReplayConvertor());
     RedisStrictCommand<Long> EVAL_INTEGER = new RedisStrictCommand<Long>("EVAL");
@@ -73,7 +80,7 @@ public interface RedisCommands {
 
     RedisCommand<Boolean> HSET = new RedisCommand<Boolean>("HSET", new BooleanReplayConvertor(), 2, ValueType.MAP);
     RedisStrictCommand<String> HINCRBYFLOAT = new RedisStrictCommand<String>("HINCRBYFLOAT");
-    RedisCommand<MapScanResult<Object, Object>> HSCAN = new RedisCommand<MapScanResult<Object, Object>>("HSCAN", new MapScanResultReplayDecoder(), ValueType.MAP);
+    RedisCommand<MapScanResult<Object, Object>> HSCAN = new RedisCommand<MapScanResult<Object, Object>>("HSCAN", new NestedMultiDecoder(new ObjectMapReplayDecoder(), new MapScanResultReplayDecoder2()), ValueType.MAP);
     RedisCommand<Map<Object, Object>> HGETALL = new RedisCommand<Map<Object, Object>>("HGETALL", new ObjectMapReplayDecoder(), ValueType.MAP);
     RedisCommand<List<Object>> HVALS = new RedisCommand<List<Object>>("HVALS", new ObjectListReplayDecoder(), ValueType.MAP_VALUE);
     RedisCommand<Boolean> HEXISTS = new RedisCommand<Boolean>("HEXISTS", new BooleanReplayConvertor(), 2, ValueType.MAP_KEY);

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

@@ -1,18 +0,0 @@
-package org.redisson.client.protocol.decoder;
-
-import org.redisson.client.protocol.Decoder;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.util.CharsetUtil;
-
-public class BooleanReplayDecoder implements Decoder<Boolean> {
-
-    @Override
-    public Boolean decode(ByteBuf buf) {
-        if (buf == null) {
-            return false;
-        }
-        return "OK".equals(buf.toString(CharsetUtil.UTF_8));
-    }
-
-}

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

@@ -0,0 +1,32 @@
+package org.redisson.client.protocol.decoder;
+
+import java.util.List;
+import java.util.Map;
+
+import com.lambdaworks.redis.output.MapScanResult;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.CharsetUtil;
+
+public class MapScanResultReplayDecoder2 implements MultiDecoder<MapScanResult<Object, Object>> {
+
+    public MultiDecoder<?> get() {
+        return (MultiDecoder<?>) this;
+    }
+
+    @Override
+    public Object decode(ByteBuf buf) {
+        return Long.valueOf(buf.toString(CharsetUtil.UTF_8));
+    }
+
+    @Override
+    public MapScanResult<Object, Object> decode(List<Object> parts) {
+        return new MapScanResult<Object, Object>((Long)parts.get(0), (Map<Object, Object>)parts.get(1));
+    }
+
+    @Override
+    public boolean isApplicable(int paramNum) {
+        return paramNum == 0;
+    }
+
+}

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

@@ -0,0 +1,58 @@
+package org.redisson.client.protocol.decoder;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+
+public class NestedMultiDecoder<T> implements MultiDecoder<Object> {
+
+    private final MultiDecoder<Object> firstDecoder;
+    private final MultiDecoder<Object> secondDecoder;
+
+    private Deque<MultiDecoder<?>> iterator;
+    private Deque<MultiDecoder<?>> flipIterator;
+
+    public NestedMultiDecoder(MultiDecoder<Object> firstDecoder, MultiDecoder<Object> secondDecoder) {
+        this.firstDecoder = firstDecoder;
+        this.secondDecoder = secondDecoder;
+
+        init(firstDecoder, secondDecoder);
+    }
+
+    private void init(MultiDecoder<Object> firstDecoder, MultiDecoder<Object> secondDecoder) {
+        iterator = new ArrayDeque<MultiDecoder<?>>(Arrays.asList(firstDecoder, secondDecoder));
+        flipIterator = new ArrayDeque<MultiDecoder<?>>(Arrays.asList(firstDecoder, secondDecoder, firstDecoder));
+    }
+
+    @Override
+    public Object decode(ByteBuf buf) throws IOException {
+        return flipIterator.peek().decode(buf);
+    }
+
+    @Override
+    public MultiDecoder<?> get() {
+        return this;
+    }
+
+    @Override
+    public boolean isApplicable(int paramNum) {
+        if (paramNum == 0) {
+            flipIterator.poll();
+            if (flipIterator.isEmpty()) {
+                init(firstDecoder, secondDecoder);
+                flipIterator.poll();
+            }
+        }
+        return flipIterator.peek().isApplicable(paramNum);
+    }
+
+    @Override
+    public Object decode(List<Object> parts) {
+        return iterator.poll().decode(parts);
+    }
+
+}

src/main/java/org/redisson/client/protocol/pubsub/PubSubMessageDecoder.java

@@ -33,7 +33,7 @@ public class PubSubMessageDecoder implements MultiDecoder<Object> {
 
     @Override
     public PubSubMessage decode(List<Object> parts) {
-        return new PubSubMessage(parts.get(1).toString(), parts.get(2).toString());
+        return new PubSubMessage(parts.get(1).toString(), parts.get(2));
     }
 
     @Override