refactoring

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

@@ -57,7 +57,6 @@ 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;
@@ -71,6 +70,7 @@ 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.StreamIdListDecoder;
+import org.redisson.client.protocol.decoder.StreamObjectMapReplayDecoder;
 import org.redisson.client.protocol.decoder.StreamResultDecoder;
 import org.redisson.client.protocol.decoder.StringDataDecoder;
 import org.redisson.client.protocol.decoder.StringListReplayDecoder;
@@ -329,8 +329,8 @@ public interface RedisCommands {
     RedisCommand<Map<String, Map<StreamMessageId, Map<Object, Object>>>> XRANGE = new RedisCommand<Map<String, Map<StreamMessageId, Map<Object, Object>>>>("XRANGE",
             new ListMultiDecoder(
                     new ObjectDecoder(new StreamIdDecoder()),
-                    new ObjectMapReplayDecoder(),
-                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET),
+                    new StreamObjectMapReplayDecoder(),
+                    new StreamObjectMapReplayDecoder(ListMultiDecoder.RESET),
                     new ObjectMapJoinDecoder()));
 
     RedisCommand<Map<String, Map<StreamMessageId, Map<Object, Object>>>> XREVRANGE = 
@@ -340,10 +340,10 @@ public interface RedisCommands {
             new ListMultiDecoder(
                     new ObjectDecoder(StringCodec.INSTANCE.getValueDecoder()),
                     new ObjectDecoder(new StreamIdDecoder()),
-                    new ObjectMapReplayDecoder(),
-                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
+                    new StreamObjectMapReplayDecoder(),
+                    new StreamObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
                     new ObjectMapJoinDecoder(), 
-                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX),
+                    new StreamObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX),
                     new StreamResultDecoder()));
             
     RedisCommand<Map<String, Map<StreamMessageId, Map<Object, Object>>>> XREAD_BLOCKING = new RedisCommand<Map<String, Map<StreamMessageId, Map<Object, Object>>>>("XREAD", XREAD.getReplayMultiDecoder());
@@ -352,10 +352,10 @@ public interface RedisCommands {
             new ListMultiDecoder(
                     new ObjectDecoder(StringCodec.INSTANCE.getValueDecoder()),
                     new ObjectDecoder(new StreamIdDecoder()),
-                    new ObjectMapReplayDecoder(),
-                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
+                    new StreamObjectMapReplayDecoder(),
+                    new StreamObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
                     new ObjectMapJoinDecoder(), 
-                    new ObjectMapReplayDecoder(),
+                    new StreamObjectMapReplayDecoder(),
                     new StreamResultDecoder()));
     
     RedisCommand<Map<StreamMessageId, Map<Object, Object>>> XREAD_BLOCKING_SINGLE = 
@@ -365,10 +365,10 @@ public interface RedisCommands {
             new ListMultiDecoder(
                     new ObjectDecoder(StringCodec.INSTANCE.getValueDecoder()),
                     new ObjectDecoder(new StreamIdDecoder()),
-                    new ObjectMapReplayDecoder(),
-                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
+                    new StreamObjectMapReplayDecoder(),
+                    new StreamObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
                     new ObjectMapJoinDecoder(), 
-                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX),
+                    new StreamObjectMapReplayDecoder(ListMultiDecoder.RESET_INDEX),
                     new StreamResultDecoder()));
 
     RedisCommand<Map<String, Map<StreamMessageId, Map<Object, Object>>>> XREADGROUP_BLOCKING = 
@@ -378,10 +378,10 @@ public interface RedisCommands {
             new ListMultiDecoder(
                     new ObjectDecoder(StringCodec.INSTANCE.getValueDecoder()),
                     new ObjectDecoder(new StreamIdDecoder()),
-                    new ObjectMapReplayDecoder(),
-                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
+                    new StreamObjectMapReplayDecoder(),
+                    new StreamObjectMapReplayDecoder(ListMultiDecoder.RESET_1),
                     new ObjectMapJoinDecoder(), 
-                    new ObjectMapReplayDecoder(),
+                    new StreamObjectMapReplayDecoder(),
                     new StreamResultDecoder()));
 
     RedisCommand<List<StreamMessageId>> XCLAIM_IDS = new RedisCommand<List<StreamMessageId>>("XCLAIM", new StreamIdListDecoder());
@@ -389,8 +389,8 @@ public interface RedisCommands {
     RedisCommand<Map<StreamMessageId, Map<Object, Object>>> XCLAIM = new RedisCommand<Map<StreamMessageId, Map<Object, Object>>>("XCLAIM",
             new ListMultiDecoder(
                     new ObjectDecoder(new StreamIdDecoder()),
-                    new ObjectMapReplayDecoder(),
-                    new ObjectMapReplayDecoder(ListMultiDecoder.RESET),
+                    new StreamObjectMapReplayDecoder(),
+                    new StreamObjectMapReplayDecoder(ListMultiDecoder.RESET),
                     new ObjectMapJoinDecoder()));
             
     RedisCommand<Map<StreamMessageId, Map<Object, Object>>> XREADGROUP_BLOCKING_SINGLE = new RedisCommand<Map<StreamMessageId, Map<Object, Object>>>("XREADGROUP",

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

@@ -29,40 +29,19 @@ import org.redisson.client.protocol.Decoder;
  */
 public class ObjectMapReplayDecoder implements MultiDecoder<Map<Object, Object>> {
 
-    private Decoder<Object> codec;
-    
-    public ObjectMapReplayDecoder() {
-    }
-    
-    public ObjectMapReplayDecoder(Decoder<Object> codec) {
-        super();
-        this.codec = codec;
-    }
-    
     @Override
     public Map<Object, Object> decode(List<Object> parts, State state) {
-        if (parts.get(0) instanceof Map) {
-            Map<Object, Object> result = new LinkedHashMap<Object, Object>(parts.size());
-            for (int i = 0; i < parts.size(); i++) {
-                result.putAll((Map<? extends Object, ? extends Object>) parts.get(i));
+        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;
-        } 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
     public Decoder<Object> getDecoder(int paramNum, State state) {
-        if (codec != null) {
-            return codec;
-        }
         return null;
     }
 

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

@@ -0,0 +1,62 @@
+/**
+ * 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.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.redisson.client.handler.State;
+import org.redisson.client.protocol.Decoder;
+
+/**
+ * 
+ * @author Nikita Koksharov
+ *
+ */
+public class StreamObjectMapReplayDecoder extends ObjectMapReplayDecoder {
+
+    private Decoder<Object> codec;
+    
+    public StreamObjectMapReplayDecoder() {
+    }
+    
+    public StreamObjectMapReplayDecoder(Decoder<Object> codec) {
+        super();
+        this.codec = codec;
+    }
+
+    @Override
+    public Map<Object, Object> decode(List<Object> parts, State state) {
+        if (parts.get(0) instanceof Map) {
+            Map<Object, Object> result = new LinkedHashMap<Object, Object>(parts.size());
+            for (int i = 0; i < parts.size(); i++) {
+                result.putAll((Map<? extends Object, ? extends Object>) parts.get(i));
+            }
+            return result;
+        }
+        return super.decode(parts, state);
+    }
+
+    @Override
+    public Decoder<Object> getDecoder(int paramNum, State state) {
+        if (codec != null) {
+            return codec;
+        }
+        return null;
+    }
+
+}