Repository: netty/netty (Java)

Issue:
remove handler cause ByteToMessageDecoder out disorder
### remove handler cause ByteToMessageDecoder out disorder
	
	I input byte stream [1,2,3,4,5], when i read `4` then remove  ByteToMessageDecoder,
	 the out is stream [1,2,3,5,4]
	
	Here are the test cases，the cases will fail


```
 @Test
    public void testDisorder(){
        ByteToMessageDecoder decoder = new ByteToMessageDecoder() {
            int count = 0;

            //read 4 byte then remove this decoder
            @Override
            protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
                out.add(in.readBytes(1));
                if (++count >= 4) {
                    ctx.pipeline().remove(this);
                }
            }
        };

        EmbeddedChannel channel = new EmbeddedChannel(decoder);
        assertTrue(channel.writeInbound(Unpooled.wrappedBuffer(new byte[]{1, 2, 3, 4, 5})));
        assertEquals(1, ((ByteBuf) channel.readInbound()).readByte());
        assertEquals(2, ((ByteBuf) channel.readInbound()).readByte());
        assertEquals(3, ((ByteBuf) channel.readInbound()).readByte());
        assertEquals(4, ((ByteBuf) channel.readInbound()).readByte());
        assertEquals(5, ((ByteBuf) channel.readInbound()).readByte());
        assertFalse(channel.finish());
    }


```
### netty 4.1

### java8

### window10


Hints:
@normanmaurer thanks

Patch:
diff --git a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
--- a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
@@ -81,7 +81,7 @@ public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in)
             try {
                 final ByteBuf buffer;
                 if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
-                    || cumulation.refCnt() > 1 || cumulation.isReadOnly()) {
+                        || cumulation.refCnt() > 1 || cumulation.isReadOnly()) {
                     // Expand cumulation (by replace it) when either there is not more room in the buffer
                     // or if the refCnt is greater then 1 which may happen when the user use slice().retain() or
                     // duplicate().retain() or if its read-only.
@@ -507,6 +507,8 @@ final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in,
             boolean removePending = decodeState == STATE_HANDLER_REMOVED_PENDING;
             decodeState = STATE_INIT;
             if (removePending) {
+                fireChannelRead(ctx, out, out.size());
+                out.clear();
                 handlerRemoved(ctx);
             }
         }


Test patch:
diff --git a/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java b/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
--- a/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
@@ -399,4 +399,31 @@ public void read(ChannelHandlerContext ctx) throws Exception {
         }
         assertFalse(channel.finish());
     }
+
+    @Test
+    public void testDisorder() {
+        ByteToMessageDecoder decoder = new ByteToMessageDecoder() {
+            int count;
+
+            //read 4 byte then remove this decoder
+            @Override
+            protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
+                out.add(in.readByte());
+                if (++count >= 4) {
+                    ctx.pipeline().remove(this);
+                }
+            }
+        };
+        EmbeddedChannel channel = new EmbeddedChannel(decoder);
+        assertTrue(channel.writeInbound(Unpooled.wrappedBuffer(new byte[]{1, 2, 3, 4, 5})));
+        assertEquals((byte) 1,  channel.readInbound());
+        assertEquals((byte) 2,  channel.readInbound());
+        assertEquals((byte) 3,  channel.readInbound());
+        assertEquals((byte) 4,  channel.readInbound());
+        ByteBuf buffer5 = channel.readInbound();
+        assertEquals((byte) 5, buffer5.readByte());
+        assertFalse(buffer5.isReadable());
+        assertTrue(buffer5.release());
+        assertFalse(channel.finish());
+    }
 }