From df8cd44dfa7a454b84ea71db00bf8f198fd3b42d Mon Sep 17 00:00:00 2001
From: Mark Paluch <mpaluch@paluch.biz>
Date: Fri, 1 Sep 2023 09:29:22 +0200
Subject: [PATCH] Refine command outputs to capture whether a segment has been
 received instead of relying on the deserialized value state #2498

---
 .../core/output/ClaimedMessagesOutput.java    | 12 +++++++-
 .../lettuce/core/output/GenericMapOutput.java | 13 +++++++--
 .../core/output/GeoCoordinatesListOutput.java |  8 +++--
 .../output/GeoCoordinatesValueListOutput.java |  8 +++--
 .../core/output/GeoWithinListOutput.java      | 29 ++++++++++++++-----
 .../core/output/KeyValueListOutput.java       |  6 +++-
 .../lettuce/core/output/KeyValueOutput.java   |  5 +++-
 .../output/KeyValueScanStreamingOutput.java   |  6 +++-
 .../output/KeyValueScoredValueOutput.java     | 14 +++++++--
 .../core/output/KeyValueStreamingOutput.java  |  6 +++-
 .../core/output/KeyValueValueListOutput.java  |  7 +++--
 .../io/lettuce/core/output/MapOutput.java     | 18 +++++++++---
 .../io/lettuce/core/output/MapScanOutput.java |  6 +++-
 .../core/output/PendingMessageListOutput.java |  6 +++-
 .../core/output/PendingMessagesOutput.java    |  8 +++--
 .../core/output/ScoredValueListOutput.java    |  6 +++-
 .../core/output/ScoredValueOutput.java        |  6 +++-
 .../core/output/ScoredValueScanOutput.java    |  8 +++--
 .../ScoredValueScanStreamingOutput.java       | 10 +++++--
 .../output/ScoredValueStreamingOutput.java    |  8 +++--
 .../core/output/SocketAddressOutput.java      |  5 +++-
 21 files changed, 152 insertions(+), 43 deletions(-)

diff --git a/src/main/java/io/lettuce/core/output/ClaimedMessagesOutput.java b/src/main/java/io/lettuce/core/output/ClaimedMessagesOutput.java
index 71d0e99d86..8566ab0684 100644
--- a/src/main/java/io/lettuce/core/output/ClaimedMessagesOutput.java
+++ b/src/main/java/io/lettuce/core/output/ClaimedMessagesOutput.java
@@ -44,8 +44,12 @@ public class ClaimedMessagesOutput<K, V> extends CommandOutput<K, V, ClaimedMess
 
     private String id;
 
+    private boolean hasId;
+
     private K key;
 
+    private boolean hasKey;
+
     private Map<K, V> body;
 
     private boolean bodyReceived;
@@ -75,8 +79,9 @@ public void set(ByteBuffer bytes) {
             return;
         }
 
-        if (key == null) {
+        if (!hasKey) {
             bodyReceived = true;
+            hasKey = true;
 
             if (bytes == null) {
                 return;
@@ -92,6 +97,7 @@ public void set(ByteBuffer bytes) {
 
         body.put(key, bytes == null ? null : codec.decodeValue(bytes));
         key = null;
+        hasKey = false;
     }
 
     @Override
@@ -101,15 +107,19 @@ public void complete(int depth) {
             messages.add(new StreamMessage<>(stream, id, body == null ? Collections.emptyMap() : body));
             bodyReceived = false;
             key = null;
+            hasKey = false;
             body = null;
             id = null;
+            hasId = false;
         }
 
         if (depth == 2 && justId) {
             messages.add(new StreamMessage<>(stream, id, null));
             key = null;
+            hasKey = false;
             body = null;
             id = null;
+            hasId = false;
         }
 
         if (depth == 0) {
diff --git a/src/main/java/io/lettuce/core/output/GenericMapOutput.java b/src/main/java/io/lettuce/core/output/GenericMapOutput.java
index 73a783edcb..5898d06517 100644
--- a/src/main/java/io/lettuce/core/output/GenericMapOutput.java
+++ b/src/main/java/io/lettuce/core/output/GenericMapOutput.java
@@ -32,6 +32,7 @@
  */
 public class GenericMapOutput<K, V> extends CommandOutput<K, V, Map<K, Object>> {
 
+    boolean hasKey;
     private K key;
 
     public GenericMapOutput(RedisCodec<K, V> codec) {
@@ -41,14 +42,16 @@ public GenericMapOutput(RedisCodec<K, V> codec) {
     @Override
     public void set(ByteBuffer bytes) {
 
-        if (key == null) {
+        if (!hasKey) {
             key = (bytes == null) ? null : codec.decodeKey(bytes);
+            hasKey = true;
             return;
         }
 
         Object value = (bytes == null) ? null : codec.decodeValue(bytes);
         output.put(key, value);
         key = null;
+        hasKey = false;
     }
 
     @Override
@@ -60,27 +63,31 @@ public void setBigNumber(ByteBuffer bytes) {
     @SuppressWarnings("unchecked")
     public void set(long integer) {
 
-        if (key == null) {
+        if (!hasKey) {
             key = (K) Long.valueOf(integer);
+            hasKey = true;
             return;
         }
 
         V value = (V) Long.valueOf(integer);
         output.put(key, value);
         key = null;
+        hasKey = false;
     }
 
     @Override
     public void set(double number) {
 
-        if (key == null) {
+        if (!hasKey) {
             key = (K) Double.valueOf(number);
+            hasKey = true;
             return;
         }
 
         Object value = Double.valueOf(number);
         output.put(key, value);
         key = null;
+        hasKey = false;
     }
 
     @Override
diff --git a/src/main/java/io/lettuce/core/output/GeoCoordinatesListOutput.java b/src/main/java/io/lettuce/core/output/GeoCoordinatesListOutput.java
index 414b8c4274..1e2b5632f2 100644
--- a/src/main/java/io/lettuce/core/output/GeoCoordinatesListOutput.java
+++ b/src/main/java/io/lettuce/core/output/GeoCoordinatesListOutput.java
@@ -15,7 +15,7 @@
  */
 package io.lettuce.core.output;
 
-import static java.lang.Double.parseDouble;
+import static java.lang.Double.*;
 
 import java.nio.ByteBuffer;
 import java.util.Collections;
@@ -35,6 +35,8 @@ public class GeoCoordinatesListOutput<K, V> extends CommandOutput<K, V, List<Geo
 
     private Double x;
 
+    boolean hasX;
+
     private boolean initialized;
 
     private Subscriber<GeoCoordinates> subscriber;
@@ -59,13 +61,15 @@ public void set(ByteBuffer bytes) {
     @Override
     public void set(double number) {
 
-        if (x == null) {
+        if (!hasX) {
             x = number;
+            hasX = true;
             return;
         }
 
         subscriber.onNext(output, new GeoCoordinates(x, number));
         x = null;
+        hasX = false;
     }
 
     @Override
diff --git a/src/main/java/io/lettuce/core/output/GeoCoordinatesValueListOutput.java b/src/main/java/io/lettuce/core/output/GeoCoordinatesValueListOutput.java
index 66015e7960..4c001cd4e9 100644
--- a/src/main/java/io/lettuce/core/output/GeoCoordinatesValueListOutput.java
+++ b/src/main/java/io/lettuce/core/output/GeoCoordinatesValueListOutput.java
@@ -15,7 +15,7 @@
  */
 package io.lettuce.core.output;
 
-import static java.lang.Double.parseDouble;
+import static java.lang.Double.*;
 
 import java.nio.ByteBuffer;
 import java.util.Collections;
@@ -34,6 +34,8 @@
 public class GeoCoordinatesValueListOutput<K, V> extends CommandOutput<K, V, List<Value<GeoCoordinates>>>
         implements StreamingOutput<Value<GeoCoordinates>> {
 
+    boolean hasX;
+
     private Double x;
 
     private boolean initialized;
@@ -60,13 +62,15 @@ public void set(ByteBuffer bytes) {
     @Override
     public void set(double number) {
 
-        if (x == null) {
+        if (!hasX) {
             x = number;
+            hasX = true;
             return;
         }
 
         subscriber.onNext(output, Value.fromNullable(new GeoCoordinates(x, number)));
         x = null;
+        hasX = false;
     }
 
     @Override
diff --git a/src/main/java/io/lettuce/core/output/GeoWithinListOutput.java b/src/main/java/io/lettuce/core/output/GeoWithinListOutput.java
index 61d202834f..fc3039e88d 100644
--- a/src/main/java/io/lettuce/core/output/GeoWithinListOutput.java
+++ b/src/main/java/io/lettuce/core/output/GeoWithinListOutput.java
@@ -15,7 +15,7 @@
  */
 package io.lettuce.core.output;
 
-import static java.lang.Double.parseDouble;
+import static java.lang.Double.*;
 
 import java.nio.ByteBuffer;
 import java.util.List;
@@ -35,19 +35,25 @@ public class GeoWithinListOutput<K, V> extends CommandOutput<K, V, List<GeoWithi
 
     private V member;
 
+    private boolean hasMember;
+
     private Double distance;
 
+    private boolean hasDistance;
+
     private Long geohash;
 
     private GeoCoordinates coordinates;
 
     private Double x;
 
-    private boolean withDistance;
+    private boolean hasX;
+
+    private final boolean withDistance;
 
-    private boolean withHash;
+    private final boolean withHash;
 
-    private boolean withCoordinates;
+    private final boolean withCoordinates;
 
     private Subscriber<GeoWithin<V>> subscriber;
 
@@ -62,8 +68,9 @@ public GeoWithinListOutput(RedisCodec<K, V> codec, boolean withDistance, boolean
     @Override
     public void set(long integer) {
 
-        if (member == null) {
+        if (!hasMember) {
             member = (V) (Long) integer;
+            hasMember = true;
             return;
         }
 
@@ -75,8 +82,9 @@ public void set(long integer) {
     @Override
     public void set(ByteBuffer bytes) {
 
-        if (member == null) {
+        if (!hasMember) {
             member = codec.decodeValue(bytes);
+            hasMember = true;
             return;
         }
 
@@ -88,15 +96,17 @@ public void set(ByteBuffer bytes) {
     public void set(double number) {
 
         if (withDistance) {
-            if (distance == null) {
+            if (!hasDistance) {
                 distance = number;
+                hasDistance = true;
                 return;
             }
         }
 
         if (withCoordinates) {
-            if (x == null) {
+            if (!hasX) {
                 x = number;
+                hasX = true;
                 return;
             }
 
@@ -111,10 +121,13 @@ public void complete(int depth) {
             subscriber.onNext(output, new GeoWithin<V>(member, distance, geohash, coordinates));
 
             member = null;
+            hasMember = false;
             distance = null;
+            hasDistance = false;
             geohash = null;
             coordinates = null;
             x = null;
+            hasX = false;
         }
     }
 
diff --git a/src/main/java/io/lettuce/core/output/KeyValueListOutput.java b/src/main/java/io/lettuce/core/output/KeyValueListOutput.java
index 02896e2a38..bdceb51a59 100644
--- a/src/main/java/io/lettuce/core/output/KeyValueListOutput.java
+++ b/src/main/java/io/lettuce/core/output/KeyValueListOutput.java
@@ -46,6 +46,8 @@ public class KeyValueListOutput<K, V> extends CommandOutput<K, V, List<KeyValue<
 
     private K key;
 
+    private boolean hasKey;
+
     public KeyValueListOutput(RedisCodec<K, V> codec) {
         super(codec, Collections.emptyList());
         setSubscriber(ListSubscriber.instance());
@@ -62,13 +64,15 @@ public KeyValueListOutput(RedisCodec<K, V> codec, Iterable<K> keys) {
     public void set(ByteBuffer bytes) {
 
         if (keys == null) {
-            if (key == null) {
+            if (!hasKey) {
                 key = codec.decodeKey(bytes);
+                hasKey = true;
                 return;
             }
 
             K key = this.key;
             this.key = null;
+            this.hasKey = false;
             subscriber.onNext(output, KeyValue.fromNullable(key, bytes == null ? null : codec.decodeValue(bytes)));
 
         } else {
diff --git a/src/main/java/io/lettuce/core/output/KeyValueOutput.java b/src/main/java/io/lettuce/core/output/KeyValueOutput.java
index b9c9d49198..8c3899ac71 100644
--- a/src/main/java/io/lettuce/core/output/KeyValueOutput.java
+++ b/src/main/java/io/lettuce/core/output/KeyValueOutput.java
@@ -32,6 +32,8 @@ public class KeyValueOutput<K, V> extends CommandOutput<K, V, KeyValue<K, V>> {
 
     private K key;
 
+    private boolean hasKey;
+
     public KeyValueOutput(RedisCodec<K, V> codec) {
         super(codec, null);
     }
@@ -40,8 +42,9 @@ public KeyValueOutput(RedisCodec<K, V> codec) {
     public void set(ByteBuffer bytes) {
 
         if (bytes != null) {
-            if (key == null) {
+            if (!hasKey) {
                 key = codec.decodeKey(bytes);
+                hasKey = true;
             } else {
                 V value = codec.decodeValue(bytes);
                 output = KeyValue.fromNullable(key, value);
diff --git a/src/main/java/io/lettuce/core/output/KeyValueScanStreamingOutput.java b/src/main/java/io/lettuce/core/output/KeyValueScanStreamingOutput.java
index 5f21157b8a..da1d45068b 100644
--- a/src/main/java/io/lettuce/core/output/KeyValueScanStreamingOutput.java
+++ b/src/main/java/io/lettuce/core/output/KeyValueScanStreamingOutput.java
@@ -32,6 +32,8 @@ public class KeyValueScanStreamingOutput<K, V> extends ScanOutput<K, V, StreamSc
 
     private K key;
 
+    private boolean hasKey;
+
     private KeyValueStreamingChannel<K, V> channel;
 
     public KeyValueScanStreamingOutput(RedisCodec<K, V> codec, KeyValueStreamingChannel<K, V> channel) {
@@ -42,8 +44,9 @@ public KeyValueScanStreamingOutput(RedisCodec<K, V> codec, KeyValueStreamingChan
     @Override
     protected void setOutput(ByteBuffer bytes) {
 
-        if (key == null) {
+        if (!hasKey) {
             key = codec.decodeKey(bytes);
+            hasKey = true;
             return;
         }
 
@@ -52,6 +55,7 @@ protected void setOutput(ByteBuffer bytes) {
         channel.onKeyValue(key, value);
         output.setCount(output.getCount() + 1);
         key = null;
+        hasKey = false;
     }
 
 }
diff --git a/src/main/java/io/lettuce/core/output/KeyValueScoredValueOutput.java b/src/main/java/io/lettuce/core/output/KeyValueScoredValueOutput.java
index 133784904e..f7ca5c7a98 100644
--- a/src/main/java/io/lettuce/core/output/KeyValueScoredValueOutput.java
+++ b/src/main/java/io/lettuce/core/output/KeyValueScoredValueOutput.java
@@ -18,9 +18,9 @@
 import java.nio.ByteBuffer;
 
 import io.lettuce.core.KeyValue;
-import io.lettuce.core.internal.LettuceStrings;
 import io.lettuce.core.ScoredValue;
 import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.internal.LettuceStrings;
 
 /**
  * {@link KeyValue} encapsulating {@link ScoredValue}. See {@code BZPOPMIN}/{@code BZPOPMAX} commands.
@@ -34,8 +34,12 @@ public class KeyValueScoredValueOutput<K, V> extends CommandOutput<K, V, KeyValu
 
     private K key;
 
+    private boolean hasKey;
+
     private V value;
 
+    private boolean hasValue;
+
     public KeyValueScoredValueOutput(RedisCodec<K, V> codec) {
         super(codec, null);
     }
@@ -47,13 +51,15 @@ public void set(ByteBuffer bytes) {
             return;
         }
 
-        if (key == null) {
+        if (!hasKey) {
             key = codec.decodeKey(bytes);
+            hasKey = true;
             return;
         }
 
-        if (value == null) {
+        if (!hasValue) {
             value = codec.decodeValue(bytes);
+            hasValue = true;
             return;
         }
 
@@ -67,7 +73,9 @@ public void set(double number) {
 
         output = KeyValue.just(key, ScoredValue.just(number, value));
         key = null;
+        hasKey = false;
         value = null;
+        hasValue = false;
     }
 
 }
diff --git a/src/main/java/io/lettuce/core/output/KeyValueStreamingOutput.java b/src/main/java/io/lettuce/core/output/KeyValueStreamingOutput.java
index 7bbb4dde84..b3726c8417 100644
--- a/src/main/java/io/lettuce/core/output/KeyValueStreamingOutput.java
+++ b/src/main/java/io/lettuce/core/output/KeyValueStreamingOutput.java
@@ -35,6 +35,8 @@ public class KeyValueStreamingOutput<K, V> extends CommandOutput<K, V, Long> {
 
     private K key;
 
+    private boolean hasKey;
+
     private KeyValueStreamingChannel<K, V> channel;
 
     public KeyValueStreamingOutput(RedisCodec<K, V> codec, KeyValueStreamingChannel<K, V> channel) {
@@ -52,8 +54,9 @@ public KeyValueStreamingOutput(RedisCodec<K, V> codec, KeyValueStreamingChannel<
     public void set(ByteBuffer bytes) {
 
         if (keys == null) {
-            if (key == null) {
+            if (!hasKey) {
                 key = codec.decodeKey(bytes);
+                hasKey = true;
                 return;
             }
         } else {
@@ -67,6 +70,7 @@ public void set(ByteBuffer bytes) {
         channel.onKeyValue(key, value);
         output = output.longValue() + 1;
         key = null;
+        hasKey = false;
     }
 
 }
diff --git a/src/main/java/io/lettuce/core/output/KeyValueValueListOutput.java b/src/main/java/io/lettuce/core/output/KeyValueValueListOutput.java
index afacb670b4..3941ba0cfe 100644
--- a/src/main/java/io/lettuce/core/output/KeyValueValueListOutput.java
+++ b/src/main/java/io/lettuce/core/output/KeyValueValueListOutput.java
@@ -34,6 +34,8 @@ public class KeyValueValueListOutput<K, V> extends CommandOutput<K, V, KeyValue<
 
     private K key;
 
+    private boolean hasKey;
+
     private List<V> values = Collections.emptyList();
 
     public KeyValueValueListOutput(RedisCodec<K, V> codec) {
@@ -44,8 +46,9 @@ public KeyValueValueListOutput(RedisCodec<K, V> codec) {
     public void set(ByteBuffer bytes) {
 
         if (bytes != null) {
-            if (key == null) {
+            if (!hasKey) {
                 key = codec.decodeKey(bytes);
+                hasKey = true;
             } else {
                 V value = codec.decodeValue(bytes);
                 values.add(value);
@@ -60,7 +63,7 @@ public void multi(int count) {
 
     @Override
     public void complete(int depth) {
-        if (depth == 0 && key != null) {
+        if (depth == 0 && hasKey) {
             output = KeyValue.just(key, values);
         }
     }
diff --git a/src/main/java/io/lettuce/core/output/MapOutput.java b/src/main/java/io/lettuce/core/output/MapOutput.java
index 5b1d60a828..e9fe10bb4c 100644
--- a/src/main/java/io/lettuce/core/output/MapOutput.java
+++ b/src/main/java/io/lettuce/core/output/MapOutput.java
@@ -37,6 +37,8 @@ public class MapOutput<K, V> extends CommandOutput<K, V, Map<K, V>> {
 
     private K key;
 
+    private boolean hasKey;
+
     public MapOutput(RedisCodec<K, V> codec) {
         super(codec, Collections.emptyMap());
     }
@@ -44,54 +46,62 @@ public MapOutput(RedisCodec<K, V> codec) {
     @Override
     public void set(ByteBuffer bytes) {
 
-        if (key == null) {
+        if (!hasKey) {
             key = (bytes == null) ? null : codec.decodeKey(bytes);
+            hasKey = true;
             return;
         }
 
         V value = (bytes == null) ? null : codec.decodeValue(bytes);
         output.put(key, value);
         key = null;
+        hasKey = false;
     }
 
     @Override
     @SuppressWarnings("unchecked")
     public void set(long integer) {
 
-        if (key == null) {
+        if (!hasKey) {
             key = (K) Long.valueOf(integer);
+            hasKey = true;
             return;
         }
 
         V value = (V) Long.valueOf(integer);
         output.put(key, value);
         key = null;
+        hasKey = false;
     }
 
     @Override
     public void set(double number) {
 
-        if (key == null) {
+        if (!hasKey) {
             key = (K) Double.valueOf(number);
+            hasKey = true;
             return;
         }
 
         V value = (V) Double.valueOf(number);
         output.put(key, value);
         key = null;
+        hasKey = false;
     }
 
     @Override
     public void set(boolean flag) {
 
-        if (key == null) {
+        if (!hasKey) {
             key = (K) Boolean.valueOf(flag);
+            hasKey = true;
             return;
         }
 
         V value = (V) Boolean.valueOf(flag);
         output.put(key, value);
         key = null;
+        hasKey = false;
     }
 
     @Override
diff --git a/src/main/java/io/lettuce/core/output/MapScanOutput.java b/src/main/java/io/lettuce/core/output/MapScanOutput.java
index 4d36b001b4..e33ce3ecda 100644
--- a/src/main/java/io/lettuce/core/output/MapScanOutput.java
+++ b/src/main/java/io/lettuce/core/output/MapScanOutput.java
@@ -31,6 +31,8 @@ public class MapScanOutput<K, V> extends ScanOutput<K, V, MapScanCursor<K, V>> {
 
     private K key;
 
+    private boolean hasKey;
+
     public MapScanOutput(RedisCodec<K, V> codec) {
         super(codec, new MapScanCursor<K, V>());
     }
@@ -38,14 +40,16 @@ public MapScanOutput(RedisCodec<K, V> codec) {
     @Override
     protected void setOutput(ByteBuffer bytes) {
 
-        if (key == null) {
+        if (!hasKey) {
             key = codec.decodeKey(bytes);
+            hasKey = true;
             return;
         }
 
         V value = (bytes == null) ? null : codec.decodeValue(bytes);
         output.getMap().put(key, value);
         key = null;
+        hasKey = false;
     }
 
 }
diff --git a/src/main/java/io/lettuce/core/output/PendingMessageListOutput.java b/src/main/java/io/lettuce/core/output/PendingMessageListOutput.java
index f483e1b477..0c2562d4c1 100644
--- a/src/main/java/io/lettuce/core/output/PendingMessageListOutput.java
+++ b/src/main/java/io/lettuce/core/output/PendingMessageListOutput.java
@@ -43,6 +43,8 @@ public class PendingMessageListOutput<K, V> extends CommandOutput<K, V, List<Pen
 
     private String consumer;
 
+    private boolean hasConsumer;
+
     private Long msSinceLastDelivery;
 
     public PendingMessageListOutput(RedisCodec<K, V> codec) {
@@ -58,8 +60,9 @@ public void set(ByteBuffer bytes) {
             return;
         }
 
-        if (consumer == null) {
+        if (!hasConsumer) {
             consumer = StringCodec.UTF8.decodeKey(bytes);
+            hasConsumer = true;
             return;
         }
 
@@ -77,6 +80,7 @@ public void set(long integer) {
         PendingMessage message = new PendingMessage(messageId, consumer, msSinceLastDelivery, integer);
         messageId = null;
         consumer = null;
+        hasConsumer = false;
         msSinceLastDelivery = null;
         subscriber.onNext(output, message);
     }
diff --git a/src/main/java/io/lettuce/core/output/PendingMessagesOutput.java b/src/main/java/io/lettuce/core/output/PendingMessagesOutput.java
index 8133c1e1d7..e5d1c78fcf 100644
--- a/src/main/java/io/lettuce/core/output/PendingMessagesOutput.java
+++ b/src/main/java/io/lettuce/core/output/PendingMessagesOutput.java
@@ -42,6 +42,8 @@ public class PendingMessagesOutput<K, V> extends CommandOutput<K, V, PendingMess
 
     private String consumer;
 
+    private boolean hasConsumer;
+
     private final Map<String, Long> consumerMessageCount = new LinkedHashMap<>();
 
     public PendingMessagesOutput(RedisCodec<K, V> codec) {
@@ -61,8 +63,9 @@ public void set(ByteBuffer bytes) {
             return;
         }
 
-        if (consumer == null) {
+        if (!hasConsumer) {
             consumer = StringCodec.UTF8.decodeKey(bytes);
+            hasConsumer = true;
             return;
         }
 
@@ -77,9 +80,10 @@ public void set(long integer) {
             return;
         }
 
-        if (consumer != null) {
+        if (hasConsumer) {
             consumerMessageCount.put(consumer, integer);
             consumer = null;
+            hasConsumer = false;
         }
     }
 
diff --git a/src/main/java/io/lettuce/core/output/ScoredValueListOutput.java b/src/main/java/io/lettuce/core/output/ScoredValueListOutput.java
index ccfd531ae3..17397292af 100644
--- a/src/main/java/io/lettuce/core/output/ScoredValueListOutput.java
+++ b/src/main/java/io/lettuce/core/output/ScoredValueListOutput.java
@@ -40,6 +40,8 @@ public class ScoredValueListOutput<K, V> extends CommandOutput<K, V, List<Scored
 
     private V value;
 
+    private boolean hasValue;
+
     public ScoredValueListOutput(RedisCodec<K, V> codec) {
         super(codec, Collections.emptyList());
         setSubscriber(ListSubscriber.instance());
@@ -48,8 +50,9 @@ public ScoredValueListOutput(RedisCodec<K, V> codec) {
     @Override
     public void set(ByteBuffer bytes) {
 
-        if (value == null) {
+        if (!hasValue) {
             value = codec.decodeValue(bytes);
+            hasValue = true;
             return;
         }
 
@@ -62,6 +65,7 @@ public void set(double number) {
 
         subscriber.onNext(output, ScoredValue.just(number, value));
         value = null;
+        hasValue = false;
     }
 
     @Override
diff --git a/src/main/java/io/lettuce/core/output/ScoredValueOutput.java b/src/main/java/io/lettuce/core/output/ScoredValueOutput.java
index 768de5f5fd..20dc747a33 100644
--- a/src/main/java/io/lettuce/core/output/ScoredValueOutput.java
+++ b/src/main/java/io/lettuce/core/output/ScoredValueOutput.java
@@ -33,6 +33,8 @@ public class ScoredValueOutput<K, V> extends CommandOutput<K, V, ScoredValue<V>>
 
     private V value;
 
+    private boolean hasValue;
+
     public ScoredValueOutput(RedisCodec<K, V> codec) {
         super(codec, ScoredValue.empty());
     }
@@ -44,8 +46,9 @@ public void set(ByteBuffer bytes) {
             return;
         }
 
-        if (value == null) {
+        if (!hasValue) {
             value = codec.decodeValue(bytes);
+            hasValue = true;
             return;
         }
 
@@ -57,6 +60,7 @@ public void set(ByteBuffer bytes) {
     public void set(double number) {
         output = ScoredValue.just(number, value);
         value = null;
+        hasValue = false;
     }
 
 }
diff --git a/src/main/java/io/lettuce/core/output/ScoredValueScanOutput.java b/src/main/java/io/lettuce/core/output/ScoredValueScanOutput.java
index 37966e247e..cb9a986e5f 100644
--- a/src/main/java/io/lettuce/core/output/ScoredValueScanOutput.java
+++ b/src/main/java/io/lettuce/core/output/ScoredValueScanOutput.java
@@ -33,6 +33,8 @@ public class ScoredValueScanOutput<K, V> extends ScanOutput<K, V, ScoredValueSca
 
     private V value;
 
+    private boolean hasValue;
+
     public ScoredValueScanOutput(RedisCodec<K, V> codec) {
         super(codec, new ScoredValueScanCursor<V>());
     }
@@ -40,8 +42,9 @@ public ScoredValueScanOutput(RedisCodec<K, V> codec) {
     @Override
     protected void setOutput(ByteBuffer bytes) {
 
-        if (value == null) {
+        if (!hasValue) {
             value = codec.decodeValue(bytes);
+            hasValue = true;
             return;
         }
 
@@ -51,10 +54,11 @@ protected void setOutput(ByteBuffer bytes) {
 
     @Override
     public void set(double number) {
-        if (value != null) {
+        if (hasValue) {
             output.getValues().add(ScoredValue.just(number, value));
         }
         value = null;
+        hasValue = false;
     }
 
 }
diff --git a/src/main/java/io/lettuce/core/output/ScoredValueScanStreamingOutput.java b/src/main/java/io/lettuce/core/output/ScoredValueScanStreamingOutput.java
index 89579f9275..b43ac97bf9 100644
--- a/src/main/java/io/lettuce/core/output/ScoredValueScanStreamingOutput.java
+++ b/src/main/java/io/lettuce/core/output/ScoredValueScanStreamingOutput.java
@@ -35,6 +35,8 @@ public class ScoredValueScanStreamingOutput<K, V> extends ScanOutput<K, V, Strea
 
     private V value;
 
+    private boolean hasValue;
+
     public ScoredValueScanStreamingOutput(RedisCodec<K, V> codec, ScoredValueStreamingChannel<V> channel) {
         super(codec, new StreamScanCursor());
         this.channel = channel;
@@ -42,8 +44,9 @@ public ScoredValueScanStreamingOutput(RedisCodec<K, V> codec, ScoredValueStreami
 
     @Override
     protected void setOutput(ByteBuffer bytes) {
-        if (value == null) {
+        if (!hasValue) {
             value = codec.decodeValue(bytes);
+            hasValue = true;
             return;
         }
 
@@ -54,12 +57,13 @@ protected void setOutput(ByteBuffer bytes) {
     @Override
     public void set(double number) {
 
-        if (value != null) {
+        if (hasValue) {
             channel.onValue(ScoredValue.just(number, value));
         }
 
-        value = null;
         output.setCount(output.getCount() + 1);
+        value = null;
+        hasValue = false;
     }
 
 }
diff --git a/src/main/java/io/lettuce/core/output/ScoredValueStreamingOutput.java b/src/main/java/io/lettuce/core/output/ScoredValueStreamingOutput.java
index 878d41e5b4..00c9d5988b 100644
--- a/src/main/java/io/lettuce/core/output/ScoredValueStreamingOutput.java
+++ b/src/main/java/io/lettuce/core/output/ScoredValueStreamingOutput.java
@@ -32,6 +32,8 @@ public class ScoredValueStreamingOutput<K, V> extends CommandOutput<K, V, Long>
 
     private V value;
 
+    private boolean hasValue;
+
     private final ScoredValueStreamingChannel<V> channel;
 
     public ScoredValueStreamingOutput(RedisCodec<K, V> codec, ScoredValueStreamingChannel<V> channel) {
@@ -42,8 +44,9 @@ public ScoredValueStreamingOutput(RedisCodec<K, V> codec, ScoredValueStreamingCh
     @Override
     public void set(ByteBuffer bytes) {
 
-        if (value == null) {
+        if (!hasValue) {
             value = codec.decodeValue(bytes);
+            hasValue = true;
             return;
         }
 
@@ -55,8 +58,9 @@ public void set(ByteBuffer bytes) {
     public void set(double number) {
 
         channel.onValue(ScoredValue.just(number, value));
-        value = null;
         output = output.longValue() + 1;
+        value = null;
+        hasValue = false;
     }
 
 }
diff --git a/src/main/java/io/lettuce/core/output/SocketAddressOutput.java b/src/main/java/io/lettuce/core/output/SocketAddressOutput.java
index 8c18bfaec2..0c08ac9915 100644
--- a/src/main/java/io/lettuce/core/output/SocketAddressOutput.java
+++ b/src/main/java/io/lettuce/core/output/SocketAddressOutput.java
@@ -31,6 +31,8 @@ public class SocketAddressOutput<K, V> extends CommandOutput<K, V, SocketAddress
 
     private String hostname;
 
+    private boolean hasHostname;
+
     public SocketAddressOutput(RedisCodec<K, V> codec) {
         super(codec, null);
     }
@@ -38,8 +40,9 @@ public SocketAddressOutput(RedisCodec<K, V> codec) {
     @Override
     public void set(ByteBuffer bytes) {
 
-        if (hostname == null) {
+        if (!hasHostname) {
             hostname = decodeAscii(bytes);
+            hasHostname = true;
             return;
         }