MultiDecoder empty result handling fixed

src/main/java/org/redisson/RedissonGeo.java

@@ -45,12 +45,19 @@ import io.netty.util.concurrent.Future;
 
 public class RedissonGeo<V> extends RedissonExpirable implements RGeo<V> {
 
+    MultiDecoder<Map<Object, Object>> postitionDecoder;
+    MultiDecoder<Map<Object, Object>> distanceDecoder;
+    
     public RedissonGeo(CommandAsyncExecutor connectionManager, String name) {
         super(connectionManager, name);
+        postitionDecoder = new NestedMultiDecoder(new GeoPositionDecoder(), new GeoDistanceDecoder(codec), new GeoMapReplayDecoder(), true);
+        distanceDecoder = new FlatNestedMultiDecoder(new GeoDistanceDecoder(codec), new GeoMapReplayDecoder(), true);
     }
     
     public RedissonGeo(Codec codec, CommandAsyncExecutor connectionManager, String name) {
         super(codec, connectionManager, name);
+        postitionDecoder = new NestedMultiDecoder(new GeoPositionDecoder(), new GeoDistanceDecoder(codec), new GeoMapReplayDecoder(), true);
+        distanceDecoder = new FlatNestedMultiDecoder(new GeoDistanceDecoder(codec), new GeoMapReplayDecoder(), true);
     }
 
     @Override
@@ -119,7 +126,7 @@ public class RedissonGeo<V> extends RedissonExpirable implements RGeo<V> {
         params.add(getName());
         params.addAll(Arrays.asList(members));
         
-        MultiDecoder<Map<Object, Object>> decoder = new NestedMultiDecoder(new GeoPositionDecoder(), new GeoPositionMapDecoder(params));
+        MultiDecoder<Map<Object, Object>> decoder = new NestedMultiDecoder(new GeoPositionDecoder(), new GeoPositionMapDecoder(params), true);
         RedisCommand<Map<Object, Object>> command = new RedisCommand<Map<Object, Object>>("GEOPOS", decoder, 2, ValueType.OBJECTS);
         return commandExecutor.readAsync(getName(), new ScoredCodec(codec), command, params.toArray());
     }
@@ -141,8 +148,7 @@ public class RedissonGeo<V> extends RedissonExpirable implements RGeo<V> {
     
     @Override
     public Future<Map<V, Double>> radiusWithDistanceAsync(double longitude, double latitude, double radius, GeoUnit geoUnit) {
-        MultiDecoder<Map<Object, Object>> decoder = new FlatNestedMultiDecoder(new GeoDistanceDecoder(codec), new GeoMapReplayDecoder());
+        RedisCommand<Map<Object, Object>> command = new RedisCommand<Map<Object, Object>>("GEORADIUS", distanceDecoder);
-        RedisCommand<Map<Object, Object>> command = new RedisCommand<Map<Object, Object>>("GEORADIUS", decoder);
         return commandExecutor.readAsync(getName(), codec, command, getName(), convert(longitude), convert(latitude), radius, geoUnit, "WITHDIST");
     }
 
@@ -153,8 +159,7 @@ public class RedissonGeo<V> extends RedissonExpirable implements RGeo<V> {
     
     @Override
     public Future<Map<V, GeoPosition>> radiusWithPositionAsync(double longitude, double latitude, double radius, GeoUnit geoUnit) {
-        MultiDecoder<Map<Object, Object>> decoder = new NestedMultiDecoder(new GeoPositionDecoder(), new GeoDistanceDecoder(codec), new GeoMapReplayDecoder());
+        RedisCommand<Map<Object, Object>> command = new RedisCommand<Map<Object, Object>>("GEORADIUS", postitionDecoder);
-        RedisCommand<Map<Object, Object>> command = new RedisCommand<Map<Object, Object>>("GEORADIUS", decoder);
         return commandExecutor.readAsync(getName(), codec, command, getName(), convert(longitude), convert(latitude), radius, geoUnit, "WITHCOORD");
     }
 
@@ -175,8 +180,7 @@ public class RedissonGeo<V> extends RedissonExpirable implements RGeo<V> {
     
     @Override
     public Future<Map<V, Double>> radiusWithDistanceAsync(V member, double radius, GeoUnit geoUnit) {
-        MultiDecoder<Map<Object, Object>> decoder = new FlatNestedMultiDecoder(new GeoDistanceDecoder(codec), new GeoMapReplayDecoder());
+        RedisCommand command = new RedisCommand("GEORADIUSBYMEMBER", distanceDecoder, 2);
-        RedisCommand command = new RedisCommand("GEORADIUSBYMEMBER", decoder, 2);
         return commandExecutor.readAsync(getName(), codec, command, getName(), member, radius, geoUnit, "WITHDIST");
     }
 
@@ -187,8 +191,7 @@ public class RedissonGeo<V> extends RedissonExpirable implements RGeo<V> {
     
     @Override
     public Future<Map<V, GeoPosition>> radiusWithPositionAsync(V member, double radius, GeoUnit geoUnit) {
-        MultiDecoder<Map<Object, Object>> decoder = new NestedMultiDecoder(new GeoPositionDecoder(), new GeoDistanceDecoder(codec), new GeoMapReplayDecoder());
+        RedisCommand<Map<Object, Object>> command = new RedisCommand<Map<Object, Object>>("GEORADIUSBYMEMBER", postitionDecoder, 2);
-        RedisCommand<Map<Object, Object>> command = new RedisCommand<Map<Object, Object>>("GEORADIUSBYMEMBER", decoder, 2);
         return commandExecutor.readAsync(getName(), codec, command, getName(), member, radius, geoUnit, "WITHCOORD");
     }
 }

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

@@ -27,6 +27,10 @@ public class FlatNestedMultiDecoder<T> extends NestedMultiDecoder {
         super(firstDecoder, secondDecoder);
     }
     
+    public FlatNestedMultiDecoder(MultiDecoder firstDecoder, MultiDecoder secondDecoder, boolean handleEmpty) {
+        super(firstDecoder, secondDecoder, handleEmpty);
+    }
+
     @Override
     public Object decode(ByteBuf buf, State state) throws IOException {
         return firstDecoder.decode(buf, state);

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

@@ -58,19 +58,27 @@ public class NestedMultiDecoder<T> implements MultiDecoder<Object> {
     protected final MultiDecoder<Object> firstDecoder;
     protected final MultiDecoder<Object> secondDecoder;
     private MultiDecoder<Object> thirdDecoder;
+    private boolean handleEmpty;
 
     public NestedMultiDecoder(MultiDecoder<Object> firstDecoder, MultiDecoder<Object> secondDecoder) {
-        this.firstDecoder = firstDecoder;
+        this(firstDecoder, secondDecoder, false);
-        this.secondDecoder = secondDecoder;
+    }
+
+    public NestedMultiDecoder(MultiDecoder<Object> firstDecoder, MultiDecoder<Object> secondDecoder, boolean handleEmpty) {
+        this(firstDecoder, secondDecoder, null, handleEmpty);
     }
     
     public NestedMultiDecoder(MultiDecoder<Object> firstDecoder, MultiDecoder<Object> secondDecoder, MultiDecoder<Object> thirdDecoder) {
+        this(firstDecoder, secondDecoder, thirdDecoder, false);
+    }
+
+    public NestedMultiDecoder(MultiDecoder<Object> firstDecoder, MultiDecoder<Object> secondDecoder, MultiDecoder<Object> thirdDecoder, boolean handleEmpty) {
         this.firstDecoder = firstDecoder;
         this.secondDecoder = secondDecoder;
         this.thirdDecoder = thirdDecoder;
+        this.handleEmpty = handleEmpty;
     }
 
-
     @Override
     public Object decode(ByteBuf buf, State state) throws IOException {
         DecoderState ds = getDecoder(state);
@@ -121,7 +129,7 @@ public class NestedMultiDecoder<T> implements MultiDecoder<Object> {
 
     @Override
     public Object decode(List<Object> parts, State state) {
-        if (parts.isEmpty() && state.getDecoderState() == null) {
+        if (parts.isEmpty() && state.getDecoderState() == null && handleEmpty) {
             MultiDecoder<?> decoder = secondDecoder;
             if (thirdDecoder != null) {
                 decoder = thirdDecoder;