Map decoding/encoding support

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

@@ -27,6 +27,7 @@ import org.redisson.client.RedisMovedException;
 import org.redisson.client.RedisPubSubConnection;
 import org.redisson.client.handler.RedisCommandsQueue.QueueCommands;
 import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.RedisCommand.ValueType;
 import org.redisson.client.protocol.pubsub.MultiDecoder;
 import org.redisson.client.protocol.pubsub.PubSubMessage;
 import org.redisson.client.protocol.pubsub.PubSubPatternMessage;
@@ -103,6 +104,14 @@ public class RedisDecoder extends ReplayingDecoder<Void> {
             }
 
             Object result = messageDecoder(data, respParts).decode(respParts);
+            handleMultiResult(data, parts, pubSubConnection, result);
+        } else {
+            throw new IllegalStateException("Can't decode replay " + (char)code);
+        }
+    }
+
+    private void handleMultiResult(RedisData<Object, Object> data, List<Object> parts,
+            RedisPubSubConnection pubSubConnection, Object result) {
         if (data != null) {
             if (Arrays.asList("PSUBSCRIBE", "SUBSCRIBE").contains(data.getCommand().getName())) {
                 for (Object param : data.getParams()) {
@@ -127,9 +136,6 @@ public class RedisDecoder extends ReplayingDecoder<Void> {
                 pubSubConnection.onMessage((PubSubPatternMessage) result);
             }
         }
-        } else {
-            throw new IllegalStateException("Can't decode replay " + (char)code);
-        }
     }
 
     private void handleResult(RedisData<Object, Object> data, List<Object> parts, Object result) {
@@ -178,7 +184,19 @@ public class RedisDecoder extends ReplayingDecoder<Void> {
             }
         }
         if (decoder == null) {
-            decoder = data.getCodec();
+            if (data.getCommand().getOutParamType() == ValueType.MAP) {
+                if (parts.size() % 2 != 0) {
+                    decoder = data.getCodec().getMapKeyDecoder();
+                } else {
+                    decoder = data.getCodec().getMapValueDecoder();
+                }
+            } else if (data.getCommand().getOutParamType() == ValueType.MAP_KEY) {
+                decoder = data.getCodec().getMapKeyDecoder();
+            } else if (data.getCommand().getOutParamType() == ValueType.MAP_VALUE) {
+                decoder = data.getCodec().getMapValueDecoder();
+            } else {
+                decoder = data.getCodec().getValueDecoder();
+            }
         }
         return decoder;
     }

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

@@ -15,7 +15,8 @@
  */
 package org.redisson.client.handler;
 
-import java.util.Arrays;
+import org.redisson.client.protocol.Encoder;
+import org.redisson.client.protocol.RedisCommand.ValueType;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
@@ -44,11 +45,15 @@ public class RedisEncoder extends MessageToByteEncoder<RedisData<Object, Object>
         }
         int i = 1;
         for (Object param : msg.getParams()) {
-            if (Arrays.binarySearch(msg.getCommand().getObjectParamIndexes(), i) != -1) {
+            Encoder encoder = msg.getCommand().getParamsEncoder();
-                writeArgument(out, msg.getCodec().encode(i, param));
+            if (msg.getCommand().getInParamIndex() == i && msg.getCommand().getInParamType() == ValueType.OBJECT) {
-            } else {
+                encoder = msg.getCodec().getValueEncoder();
-                writeArgument(out, msg.getCommand().getParamsEncoder().encode(i, param));
+            } else if (msg.getCommand().getInParamIndex() <= i && msg.getCommand().getInParamType() != ValueType.OBJECT) {
+                encoder = encoder(msg, i - msg.getCommand().getInParamIndex());
             }
+
+            writeArgument(out, encoder.encode(i, param));
+
             i++;
         }
 
@@ -56,6 +61,23 @@ public class RedisEncoder extends MessageToByteEncoder<RedisData<Object, Object>
         System.out.println(o);
     }
 
+    private Encoder encoder(RedisData<Object, Object> msg, int param) {
+        if (msg.getCommand().getInParamType() == ValueType.MAP) {
+            if (param % 2 != 0) {
+                return msg.getCodec().getMapValueEncoder();
+            } else {
+                return msg.getCodec().getMapKeyEncoder();
+            }
+        }
+        if (msg.getCommand().getInParamType() == ValueType.MAP_KEY) {
+            return msg.getCodec().getMapKeyEncoder();
+        }
+        if (msg.getCommand().getInParamType() == ValueType.MAP_KEY) {
+            return msg.getCodec().getMapValueEncoder();
+        }
+        throw new IllegalStateException();
+    }
+
     private void writeArgument(ByteBuf out, byte[] arg) {
         out.writeByte(BYTES_PREFIX);
         out.writeBytes(toChars(arg.length));

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

@@ -15,6 +15,18 @@
  */
 package org.redisson.client.protocol;
 
-public interface Codec extends Encoder, Decoder<Object> {
+public interface Codec {
+
+    Decoder<Object> getMapValueDecoder();
+
+    Encoder getMapValueEncoder();
+
+    Decoder<Object> getMapKeyDecoder();
+
+    Encoder getMapKeyEncoder();
+
+    Decoder<Object> getValueDecoder();
+
+    Encoder getValueEncoder();
 
 }

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

@@ -24,6 +24,22 @@ public class IntegerCodec implements Codec {
 
     public static final IntegerCodec INSTANCE = new IntegerCodec();
 
+    @Override
+    public Decoder<Object> getValueDecoder() {
+        return new Decoder<Object>() {
+            @Override
+            public Object decode(ByteBuf buf) {
+                if (buf == null) {
+                    return null;
+                }
+                return Integer.valueOf(buf.toString(CharsetUtil.UTF_8));
+            }
+        };
+    }
+
+    @Override
+    public Encoder getValueEncoder() {
+        return new Encoder() {
             @Override
             public byte[] encode(int paramIndex, Object in) {
                 try {
@@ -32,13 +48,27 @@ public class IntegerCodec implements Codec {
                     throw new IllegalStateException(e);
                 }
             }
+        };
+    }
 
     @Override
-    public Object decode(ByteBuf buf) {
+    public Decoder<Object> getMapValueDecoder() {
-        if (buf == null) {
+        return getValueDecoder();
-            return null;
     }
-        return Integer.valueOf(buf.toString(CharsetUtil.UTF_8));
+
+    @Override
+    public Encoder getMapValueEncoder() {
+        return getValueEncoder();
+    }
+
+    @Override
+    public Decoder<Object> getMapKeyDecoder() {
+        return getValueDecoder();
+    }
+
+    @Override
+    public Encoder getMapKeyEncoder() {
+        return getValueEncoder();
     }
 
 }

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

@@ -1,35 +0,0 @@
-package org.redisson.client.protocol;
-
-import java.io.IOException;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufInputStream;
-
-public class JsonCodec implements Codec {
-
-    public static final JsonCodec INSTANCE = new JsonCodec();
-
-    private final ObjectMapper mapper = new ObjectMapper();
-
-    @Override
-    public byte[] encode(int paramIndex, Object in) {
-        try {
-            return mapper.writeValueAsBytes(in);
-        } catch (JsonProcessingException e) {
-            throw new IllegalStateException(e);
-        }
-    }
-
-    @Override
-    public Object decode(ByteBuf buf) {
-        try {
-            return mapper.readValue(new ByteBufInputStream(buf), Object.class);
-        } catch (IOException e) {
-            throw new IllegalStateException(e);
-        }
-    }
-
-}

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

@@ -19,53 +19,103 @@ import org.redisson.client.protocol.pubsub.MultiDecoder;
 
 public class RedisCommand<R> {
 
+    public enum ValueType {OBJECT, MAP_VALUE, MAP_KEY, MAP}
+
+    private ValueType outParamType = ValueType.OBJECT;
+    private ValueType inParamType = ValueType.OBJECT;
+    private final int inParamIndex;
+
     private final String name;
     private final String subName;
-    private final int[] objectParamIndexes;
 
     private Encoder paramsEncoder = new StringParamsEncoder();
     private MultiDecoder<R> replayMultiDecoder;
     private Decoder<R> replayDecoder;
     private Convertor<R> convertor = new EmptyConvertor<R>();
 
-    public RedisCommand(String name, String subName, int ... encodeParamIndexes) {
+    public RedisCommand(String name) {
-        this(name, subName, null, null, encodeParamIndexes);
+        this(name, (String)null);
     }
 
-    public RedisCommand(String name, int ... encodeParamIndexes) {
+    public RedisCommand(String name, ValueType outParamType) {
-        this(name, null, null, null, encodeParamIndexes);
+        this(name, (String)null);
+        this.outParamType = outParamType;
     }
 
-    public RedisCommand(String name, Convertor<R> convertor, int ... encodeParamIndexes) {
+    public RedisCommand(String name, String subName) {
-        this.name = name;
+        this(name, subName, null, null, -1);
-        this.subName = null;
+    }
-        this.objectParamIndexes = encodeParamIndexes;
+
+    public RedisCommand(String name, String subName, int objectParamIndex) {
+        this(name, subName, null, null, objectParamIndex);
+    }
+
+    public RedisCommand(String name, int encodeParamIndex) {
+        this(name, null, null, null, encodeParamIndex);
+    }
+
+    public RedisCommand(String name, int encodeParamIndex, ValueType inParamType, ValueType outParamType) {
+        this(name, null, null, null, encodeParamIndex);
+        this.inParamType = inParamType;
+        this.outParamType = outParamType;
+    }
+
+
+    public RedisCommand(String name, Convertor<R> convertor, int encodeParamIndex, ValueType inParamType) {
+        this(name, null, null, null, encodeParamIndex);
         this.convertor = convertor;
+        this.inParamType = inParamType;
     }
 
-    public RedisCommand(String name, Decoder<R> reponseDecoder, int ... encodeParamIndexes) {
+    public RedisCommand(String name, Convertor<R> convertor, int encodeParamIndex) {
-        this(name, null, null, reponseDecoder, encodeParamIndexes);
+        this(name, null, null, null, encodeParamIndex);
+        this.convertor = convertor;
     }
 
-    public RedisCommand(String name, MultiDecoder<R> replayMultiDecoder, int ... encodeParamIndexes) {
+    public RedisCommand(String name, Decoder<R> reponseDecoder) {
-        this.name = name;
+        this(name, null, null, reponseDecoder, -1);
-        this.subName = null;
+    }
-        this.objectParamIndexes = encodeParamIndexes;
+
-        this.replayMultiDecoder = replayMultiDecoder;
+    public RedisCommand(String name, Decoder<R> reponseDecoder, int objectParamIndex, ValueType inParamType) {
+        this(name, null, null, reponseDecoder, objectParamIndex);
+        this.inParamType = inParamType;
+    }
+
+    public RedisCommand(String name, Decoder<R> reponseDecoder, int objectParamIndex) {
+        this(name, null, null, reponseDecoder, objectParamIndex);
+    }
+
+    public RedisCommand(String name, MultiDecoder<R> replayMultiDecoder, ValueType outParamType) {
+        this(name, replayMultiDecoder, -1);
+        this.outParamType = outParamType;
+    }
+
+    public RedisCommand(String name, MultiDecoder<R> replayMultiDecoder, int objectParamIndex, ValueType inParamType, ValueType outParamType) {
+        this(name, replayMultiDecoder, objectParamIndex);
+        this.outParamType = outParamType;
+        this.inParamType = inParamType;
+    }
+
+    public RedisCommand(String name, MultiDecoder<R> replayMultiDecoder) {
+        this(name, replayMultiDecoder, -1);
+    }
+
+    public RedisCommand(String name, MultiDecoder<R> replayMultiDecoder, int objectParamIndex) {
+        this(name, null, replayMultiDecoder, null, objectParamIndex);
     }
 
     public RedisCommand(String name, String subName, MultiDecoder<R> replayMultiDecoder,
-            int... encodeParamIndexes) {
+            int objectParamIndex) {
-        this(name, subName, replayMultiDecoder, null, encodeParamIndexes);
+        this(name, subName, replayMultiDecoder, null, objectParamIndex);
     }
 
-    public RedisCommand(String name, String subName, MultiDecoder<R> replayMultiDecoder, Decoder<R> reponseDecoder, int ... encodeParamIndexes) {
+    public RedisCommand(String name, String subName, MultiDecoder<R> replayMultiDecoder, Decoder<R> reponseDecoder, int objectParamIndex) {
         super();
         this.name = name;
         this.subName = subName;
         this.replayMultiDecoder = replayMultiDecoder;
         this.replayDecoder = reponseDecoder;
-        this.objectParamIndexes = encodeParamIndexes;
+        this.inParamIndex = objectParamIndex;
     }
 
     public String getSubName() {
@@ -80,8 +130,8 @@ public class RedisCommand<R> {
         return replayDecoder;
     }
 
-    public int[] getObjectParamIndexes() {
+    public int getInParamIndex() {
-        return objectParamIndexes;
+        return inParamIndex;
     }
 
     public MultiDecoder<R> getReplayMultiDecoder() {
@@ -96,4 +146,12 @@ public class RedisCommand<R> {
         return paramsEncoder;
     }
 
+    public ValueType getInParamType() {
+        return inParamType;
+    }
+
+    public ValueType getOutParamType() {
+        return outParamType;
+    }
+
 }

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

@@ -18,7 +18,9 @@ package org.redisson.client.protocol;
 import java.util.List;
 import java.util.Map;
 
+import org.redisson.client.protocol.RedisCommand.ValueType;
 import org.redisson.client.protocol.decoder.BooleanReplayDecoder;
+import org.redisson.client.protocol.decoder.KeyValueObjectDecoder;
 import org.redisson.client.protocol.decoder.ObjectListReplayDecoder;
 import org.redisson.client.protocol.decoder.ObjectMapReplayDecoder;
 import org.redisson.client.protocol.decoder.StringDataDecoder;
@@ -48,6 +50,7 @@ public interface RedisCommands {
     RedisStrictCommand<Long> EVAL_INTEGER = new RedisStrictCommand<Long>("EVAL");
     RedisCommand<List<Object>> EVAL_LIST = new RedisCommand<List<Object>>("EVAL", new ObjectListReplayDecoder());
     RedisCommand<Object> EVAL_OBJECT = new RedisCommand<Object>("EVAL");
+    RedisCommand<Object> EVAL_MAP_VALUE = new RedisCommand<Object>("EVAL", ValueType.MAP_VALUE);
 
     RedisStrictCommand<Long> INCR = new RedisStrictCommand<Long>("INCR");
     RedisStrictCommand<Long> INCRBY = new RedisStrictCommand<Long>("INCRBY");
@@ -61,14 +64,14 @@ public interface RedisCommands {
 
     RedisStrictCommand<List<String>> KEYS = new RedisStrictCommand<List<String>>("KEYS", new StringListReplayDecoder());
 
-    RedisCommand<Map<Object, Object>> HGETALL = new RedisCommand<Map<Object, Object>>("HGETALL", new ObjectMapReplayDecoder());
+    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());
+    RedisCommand<List<Object>> HVALS = new RedisCommand<List<Object>>("HVALS", new ObjectListReplayDecoder(), ValueType.MAP_VALUE);
-    RedisCommand<Boolean> HEXISTS = new RedisCommand<Boolean>("HEXISTS", new BooleanReplayConvertor(), 1);
+    RedisCommand<Boolean> HEXISTS = new RedisCommand<Boolean>("HEXISTS", new BooleanReplayConvertor(), 1, ValueType.MAP_KEY);
     RedisStrictCommand<Long> HLEN = new RedisStrictCommand<Long>("HLEN");
-    RedisCommand<List<Object>> HKEYS = new RedisCommand<List<Object>>("HKEYS", new ObjectListReplayDecoder());
+    RedisCommand<List<Object>> HKEYS = new RedisCommand<List<Object>>("HKEYS", new ObjectListReplayDecoder(), ValueType.MAP_KEY);
-    RedisCommand<String> HMSET = new RedisCommand<String>("HMSET", new StringReplayDecoder(), 2, 3);
+    RedisCommand<String> HMSET = new RedisCommand<String>("HMSET", new StringReplayDecoder(), 1, ValueType.MAP);
-    RedisCommand<List<Object>> HMGET = new RedisCommand<List<Object>>("HMGET", new ObjectListReplayDecoder());
+    RedisCommand<List<Object>> HMGET = new RedisCommand<List<Object>>("HMGET", new ObjectListReplayDecoder(), 1, ValueType.MAP_KEY, ValueType.MAP_VALUE);
-    RedisCommand<Object> HGET = new RedisCommand<Object>("HMGET", 2);
+    RedisCommand<Object> HGET = new RedisCommand<Object>("HGET", 1, ValueType.MAP_KEY, ValueType.MAP_VALUE);
 
     RedisStrictCommand<Boolean> DEL_BOOLEAN = new RedisStrictCommand<Boolean>("DEL", new BooleanReplayConvertor());
 

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

@@ -19,34 +19,28 @@ import org.redisson.client.protocol.pubsub.MultiDecoder;
 
 public class RedisStrictCommand<T> extends RedisCommand<T> {
 
-    public RedisStrictCommand(String name, MultiDecoder<T> replayMultiDecoder, int... encodeParamIndexes) {
+    public RedisStrictCommand(String name, MultiDecoder<T> replayMultiDecoder) {
-        super(name, replayMultiDecoder, encodeParamIndexes);
+        super(name, replayMultiDecoder);
     }
 
-    public RedisStrictCommand(String name, String subName, MultiDecoder<T> replayMultiDecoder,
+    public RedisStrictCommand(String name, String subName, MultiDecoder<T> replayMultiDecoder) {
-            int... encodeParamIndexes) {
+        super(name, subName, replayMultiDecoder, -1);
-        super(name, subName, replayMultiDecoder, encodeParamIndexes);
     }
 
-    public RedisStrictCommand(String name, int... encodeParamIndexes) {
+    public RedisStrictCommand(String name) {
-        super(name, encodeParamIndexes);
+        super(name);
     }
 
-    public RedisStrictCommand(String name, Convertor<T> convertor, int ... encodeParamIndexes) {
+    public RedisStrictCommand(String name, Convertor<T> convertor) {
-        super(name, convertor, encodeParamIndexes);
+        super(name, convertor, -1);
     }
 
-    public RedisStrictCommand(String name, String subName, Decoder<T> reponseDecoder,
+    public RedisStrictCommand(String name, String subName, Decoder<T> reponseDecoder) {
-            int... encodeParamIndexes) {
+        super(name, subName, null, reponseDecoder, -1);
-        super(name, subName, null, reponseDecoder, encodeParamIndexes);
     }
 
-    public RedisStrictCommand(String name, String subName, int... encodeParamIndexes) {
+    public RedisStrictCommand(String name, Decoder<T> reponseDecoder) {
-        super(name, subName, encodeParamIndexes);
+        super(name, reponseDecoder);
-    }
-
-    public RedisStrictCommand(String name, Decoder<T> reponseDecoder, int... encodeParamIndexes) {
-        super(name, reponseDecoder, encodeParamIndexes);
     }
 
 }

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

@@ -24,6 +24,22 @@ public class StringCodec implements Codec {
 
     public static final StringCodec INSTANCE = new StringCodec();
 
+    @Override
+    public Decoder<Object> getValueDecoder() {
+        return new Decoder<Object>() {
+            @Override
+            public Object decode(ByteBuf buf) {
+                if (buf == null) {
+                    return null;
+                }
+                return buf.toString(CharsetUtil.UTF_8);
+            }
+        };
+    }
+
+    @Override
+    public Encoder getValueEncoder() {
+        return new Encoder() {
             @Override
             public byte[] encode(int paramIndex, Object in) {
                 try {
@@ -32,13 +48,27 @@ public class StringCodec implements Codec {
                     throw new IllegalStateException(e);
                 }
             }
+        };
+    }
 
     @Override
-    public Object decode(ByteBuf buf) {
+    public Decoder<Object> getMapValueDecoder() {
-        if (buf == null) {
+        return getValueDecoder();
-            return null;
     }
-        return buf.toString(CharsetUtil.UTF_8);
+
+    @Override
+    public Encoder getMapValueEncoder() {
+        return getValueEncoder();
+    }
+
+    @Override
+    public Decoder<Object> getMapKeyDecoder() {
+        return getValueDecoder();
+    }
+
+    @Override
+    public Encoder getMapKeyEncoder() {
+        return getValueEncoder();
     }
 
 }

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

@@ -0,0 +1,82 @@
+/**
+ * Copyright 2014 Nikita Koksharov, Nickolay Borbit
+ *
+ * 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;
+
+import java.io.UnsupportedEncodingException;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.CharsetUtil;
+
+public class StringIntegerCodec implements Codec {
+
+    public static final StringIntegerCodec INSTANCE = new StringIntegerCodec();
+
+    @Override
+    public Decoder<Object> getValueDecoder() {
+        return new Decoder<Object>() {
+            @Override
+            public Object decode(ByteBuf buf) {
+                if (buf == null) {
+                    return null;
+                }
+                return buf.toString(CharsetUtil.UTF_8);
+            }
+        };
+    }
+
+    @Override
+    public Encoder getValueEncoder() {
+        return new Encoder() {
+            @Override
+            public byte[] encode(int paramIndex, Object in) {
+                try {
+                    return in.toString().getBytes("UTF-8");
+                } catch (UnsupportedEncodingException e) {
+                    throw new IllegalStateException(e);
+                }
+            }
+        };
+    }
+
+    @Override
+    public Decoder<Object> getMapValueDecoder() {
+        return new Decoder<Object>() {
+            @Override
+            public Object decode(ByteBuf buf) {
+                if (buf == null) {
+                    return null;
+                }
+                return Integer.valueOf(buf.toString(CharsetUtil.UTF_8));
+            }
+        };
+    }
+
+    @Override
+    public Encoder getMapValueEncoder() {
+        return getValueEncoder();
+    }
+
+    @Override
+    public Decoder<Object> getMapKeyDecoder() {
+        return getValueDecoder();
+    }
+
+    @Override
+    public Encoder getMapKeyEncoder() {
+        return getValueEncoder();
+    }
+
+}

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

@@ -1,4 +1,4 @@
-package org.redisson.client.protocol;
+package org.redisson.client.protocol.decoder;
 
 public class KeyValueMessage<K, V> {
 

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

@@ -1,4 +1,4 @@
-package org.redisson.client.protocol;
+package org.redisson.client.protocol.decoder;
 
 import java.util.List;