RedisDecoder infinite loop

Motivation:
RedisDecoder can get into an infinite loop while decoding bulk strings if the final \r and \n to indicate the end of content are split on ByteBuf boundaries.

Modifications:
- We should break out of the decode loop if remainingBulkLength is 0 and we don't have enough data to read EOL

Result:
No more infinite loop in RedisDecoder#decodeBulkStringContent.

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

@@ -201,7 +201,7 @@ private boolean decodeBulkStringEndOfLine(ByteBuf in, List<Object> out) throws E
     // ${expectedBulkLength}\r\n <here> {data...}\r\n
     private boolean decodeBulkStringContent(ByteBuf in, List<Object> out) throws Exception {
         final int readableBytes = in.readableBytes();
-        if (readableBytes == 0) {
+        if (readableBytes == 0 || remainingBulkLength == 0 && readableBytes < RedisConstants.EOL_LENGTH) {
             return false;
         }
 

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

@@ -27,9 +27,15 @@
 
 import java.util.List;
 
-import static io.netty.handler.codec.redis.RedisCodecTestUtil.*;
-import static org.hamcrest.CoreMatchers.*;
-import static org.junit.Assert.*;
+import static io.netty.handler.codec.redis.RedisCodecTestUtil.byteBufOf;
+import static io.netty.handler.codec.redis.RedisCodecTestUtil.bytesOf;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 
 /**
  * Verifies the correct functionality of the {@link RedisDecoder} and {@link RedisArrayAggregator}.
@@ -51,6 +57,16 @@ public void teardown() throws Exception {
         assertFalse(channel.finish());
     }
 
+    @Test
+    public void splitEOLDoesNotInfiniteLoop() throws Exception {
+        assertFalse(channel.writeInbound(byteBufOf("$6\r\nfoobar\r")));
+        assertTrue(channel.writeInbound(byteBufOf("\n")));
+
+        RedisMessage msg = channel.readInbound();
+        assertTrue(msg instanceof FullBulkStringRedisMessage);
+        ReferenceCountUtil.release(msg);
+    }
+
     @Test
     public void shouldDecodeSimpleString() {
         assertFalse(channel.writeInbound(byteBufOf("+")));