Fix decoding null bulk string of RedisDecoder

Motivation:

- `RedisBulkStringAggregator` raises errors for multiple null bulk strings.
- Null or empty bulk string has no content, but current `RedisDecoder` generates header and contents.

Modifications:

- Fix decoding null bulk string of `RedisDecoder` for `RedisBulkStringAggregator`.

Result:

- Fixes #5184.

codec-redis/src/main/java/io/netty/handler/codec/redis/BulkStringHeaderRedisMessage.java

@@ -28,6 +28,9 @@ public class BulkStringHeaderRedisMessage implements RedisMessage {
      * @param bulkStringLength follow content length.
      */
     public BulkStringHeaderRedisMessage(int bulkStringLength) {
+        if (bulkStringLength <= 0) {
+            throw new RedisCodecException("bulkStringLength: " + bulkStringLength + " (expected: > 0)");
+        }
         this.bulkStringLength = bulkStringLength;
     }
 

codec-redis/src/main/java/io/netty/handler/codec/redis/RedisBulkStringAggregator.java

@@ -93,13 +93,6 @@ public final class RedisBulkStringAggregator extends MessageAggregator<RedisMess
     @Override
     protected FullBulkStringRedisMessage beginAggregation(BulkStringHeaderRedisMessage start, ByteBuf content)
             throws Exception {
-        switch (start.bulkStringLength()) {
+        return new FullBulkStringRedisMessage(content);
-        case RedisConstants.NULL_VALUE:
-            return FullBulkStringRedisMessage.NULL_INSTANCE;
-        case 0:
-            return FullBulkStringRedisMessage.EMPTY_INSTANCE;
-        default:
-            return new FullBulkStringRedisMessage(content);
-        }
     }
 }

codec-redis/src/main/java/io/netty/handler/codec/redis/RedisDecoder.java

@@ -163,15 +163,14 @@ public final class RedisDecoder extends ByteToMessageDecoder {
                                               RedisConstants.REDIS_MESSAGE_MAX_LENGTH + ")");
             }
             remainingBulkLength = (int) length; // range(int) is already checked.
-            out.add(new BulkStringHeaderRedisMessage(remainingBulkLength));
+            return decodeBulkString(in, out);
-            return decodeBulkString(remainingBulkLength, in, out);
         default:
             throw new RedisCodecException("bad type: " + type);
         }
     }
 
-    private boolean decodeBulkString(int length, ByteBuf in, List<Object> out) throws Exception {
+    private boolean decodeBulkString(ByteBuf in, List<Object> out) throws Exception {
-        switch (length) {
+        switch (remainingBulkLength) {
         case RedisConstants.NULL_VALUE: // $-1\r\n
             out.add(FullBulkStringRedisMessage.NULL_INSTANCE);
             resetDecoder();
@@ -180,6 +179,7 @@ public final class RedisDecoder extends ByteToMessageDecoder {
             state = State.DECODE_BULK_STRING_EOL;
             return decodeBulkStringEndOfLine(in, out);
         default: // expectedBulkLength is always positive.
+            out.add(new BulkStringHeaderRedisMessage(remainingBulkLength));
             state = State.DECODE_BULK_STRING_CONTENT;
             return decodeBulkStringContent(in, out);
         }

codec-redis/src/test/java/io/netty/handler/codec/redis/RedisDecoderTest.java

@@ -153,11 +153,20 @@ public class RedisDecoderTest {
         assertFalse(channel.writeInbound(byteBufOf(Integer.toString(-1))));
         assertTrue(channel.writeInbound(byteBufOf("\r\n")));
 
-        FullBulkStringRedisMessage msg = channel.readInbound();
+        assertTrue(channel.writeInbound(byteBufOf("$")));
+        assertTrue(channel.writeInbound(byteBufOf(Integer.toString(-1))));
+        assertTrue(channel.writeInbound(byteBufOf("\r\n")));
 
-        assertThat(msg.isNull(), is(true));
+        FullBulkStringRedisMessage msg1 = channel.readInbound();
+        assertThat(msg1.isNull(), is(true));
+        ReferenceCountUtil.release(msg1);
 
-        ReferenceCountUtil.release(msg);
+        FullBulkStringRedisMessage msg2 = channel.readInbound();
+        assertThat(msg2.isNull(), is(true));
+        ReferenceCountUtil.release(msg2);
+
+        FullBulkStringRedisMessage msg3 = channel.readInbound();
+        assertThat(msg3, is(nullValue()));
     }
 
     @Test