From 9c4bfa44d97f3056a7432539a032d47f8b6b7005 Mon Sep 17 00:00:00 2001
From: Norman Maurer <nmaurer@redhat.com>
Date: Mon, 22 Apr 2013 11:04:56 +0200
Subject: [PATCH] [#1294] Make sure ByteBuf is released once written to channel

---
 .../codec/http/HttpContentDecoder.java        |  3 +-
 .../codec/http/HttpContentEncoder.java        |  3 +-
 .../handler/codec/compression/JZlibTest.java  |  6 +-
 .../compression/SnappyFramedEncoderTest.java  |  5 +-
 .../socket/SocketBufReleaseTest.java          | 87 +++++++++++++++++++
 .../socket/SocketFixedLengthEchoTest.java     |  2 +-
 .../channel/DefaultChannelHandlerContext.java |  6 +-
 7 files changed, 103 insertions(+), 9 deletions(-)
 create mode 100644 testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketBufReleaseTest.java

diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecoder.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecoder.java
index 5137c8f368..eccb9547ad 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecoder.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecoder.java
@@ -221,7 +221,8 @@ public abstract class HttpContentDecoder extends MessageToMessageDecoder<HttpObj
     }
 
     private void decode(ByteBuf in, ByteBuf out) {
-        decoder.writeInbound(in);
+        // call retain as it will be release after is written
+        decoder.writeInbound(in.retain());
         fetchDecoderOutput(out);
     }
 
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentEncoder.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentEncoder.java
index 682cc28358..b8f1e6522d 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentEncoder.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentEncoder.java
@@ -247,7 +247,8 @@ public abstract class HttpContentEncoder extends MessageToMessageCodec<HttpReque
     }
 
     private void encode(ByteBuf in, ByteBuf out) {
-        encoder.writeOutbound(in);
+        // call retain here as it will call release after its written to the channel
+        encoder.writeOutbound(in.retain());
         fetchEncoderOutput(out);
     }
 
diff --git a/codec/src/test/java/io/netty/handler/codec/compression/JZlibTest.java b/codec/src/test/java/io/netty/handler/codec/compression/JZlibTest.java
index 9eb05d8c8c..a8477f91d5 100644
--- a/codec/src/test/java/io/netty/handler/codec/compression/JZlibTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/compression/JZlibTest.java
@@ -32,7 +32,7 @@ public class JZlibTest {
         EmbeddedByteChannel chEncoder =
           new EmbeddedByteChannel(new JZlibEncoder(ZlibWrapper.ZLIB));
 
-        chEncoder.writeOutbound(data);
+        chEncoder.writeOutbound(data.copy());
         assertTrue(chEncoder.finish());
 
         byte[] deflatedData = chEncoder.readOutbound().array();
@@ -61,7 +61,7 @@ public class JZlibTest {
         EmbeddedByteChannel chEncoder =
           new EmbeddedByteChannel(new JZlibEncoder(ZlibWrapper.NONE));
 
-        chEncoder.writeOutbound(data);
+        chEncoder.writeOutbound(data.copy());
         assertTrue(chEncoder.finish());
 
         byte[] deflatedData = chEncoder.readOutbound().array();
@@ -90,7 +90,7 @@ public class JZlibTest {
         EmbeddedByteChannel chEncoder =
           new EmbeddedByteChannel(new JZlibEncoder(ZlibWrapper.GZIP));
 
-        chEncoder.writeOutbound(data);
+        chEncoder.writeOutbound(data.copy());
         assertTrue(chEncoder.finish());
 
         byte[] deflatedData = chEncoder.readOutbound().array();
diff --git a/codec/src/test/java/io/netty/handler/codec/compression/SnappyFramedEncoderTest.java b/codec/src/test/java/io/netty/handler/codec/compression/SnappyFramedEncoderTest.java
index f40274ca0b..661a99159c 100644
--- a/codec/src/test/java/io/netty/handler/codec/compression/SnappyFramedEncoderTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/compression/SnappyFramedEncoderTest.java
@@ -73,9 +73,9 @@ public class SnappyFramedEncoderTest {
             'n', 'e', 't', 't', 'y'
         });
 
-        channel.writeOutbound(in);
+        channel.writeOutbound(in.copy());
         in.readerIndex(0); // rewind the buffer to write the same data
-        channel.writeOutbound(in);
+        channel.writeOutbound(in.copy());
         assertTrue(channel.finish());
 
         ByteBuf expected = Unpooled.wrappedBuffer(new byte[] {
@@ -84,6 +84,7 @@ public class SnappyFramedEncoderTest {
              0x01, 0x09, 0x00, 0x00, 0x2d, -0x5a, -0x7e, -0x5e, 'n', 'e', 't', 't', 'y',
         });
         assertEquals(expected, channel.readOutbound());
+        in.release();
     }
 
     /**
diff --git a/testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketBufReleaseTest.java b/testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketBufReleaseTest.java
new file mode 100644
index 0000000000..1e1af13294
--- /dev/null
+++ b/testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketBufReleaseTest.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.testsuite.transport.socket;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundMessageHandlerAdapter;
+import org.junit.Test;
+
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+
+import static org.junit.Assert.assertEquals;
+
+public class SocketBufReleaseTest extends AbstractSocketTest {
+
+    @Test
+    public void testBufRelease() throws Throwable {
+        run();
+    }
+
+    public void testBufRelease(ServerBootstrap sb, Bootstrap cb) throws Throwable {
+        BufWriterHandler serverHandler = new BufWriterHandler();
+        BufWriterHandler clientHandler = new BufWriterHandler();
+
+        sb.childHandler(serverHandler);
+        cb.handler(clientHandler);
+
+        Channel sc = sb.bind().sync().channel();
+        Channel cc = cb.connect().sync().channel();
+
+        sc.close().sync();
+        cc.close().sync();
+        serverHandler.check();
+        clientHandler.check();
+    }
+
+    private static class BufWriterHandler extends ChannelInboundMessageHandlerAdapter<Object> {
+
+        private final Random random = new Random();
+        private final CountDownLatch latch = new CountDownLatch(1);
+        private ByteBuf buf;
+        @Override
+        public void channelActive(final ChannelHandlerContext ctx) throws Exception {
+            byte[] data = new byte[1024];
+            random.nextBytes(data);
+
+            buf = ctx.alloc().buffer();
+            buf.writeBytes(data);
+
+            ctx.channel().write(buf).addListener(new ChannelFutureListener() {
+                @Override
+                public void operationComplete(ChannelFuture future) throws Exception {
+                    latch.countDown();
+                }
+            });
+        }
+
+        @Override
+        public void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
+            // Discard
+        }
+
+        public void check() throws InterruptedException {
+            latch.await();
+            assertEquals(0, buf.refCnt());
+        }
+    }
+}
diff --git a/testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketFixedLengthEchoTest.java b/testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketFixedLengthEchoTest.java
index 4cf343ac2c..206406dd09 100644
--- a/testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketFixedLengthEchoTest.java
+++ b/testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketFixedLengthEchoTest.java
@@ -149,7 +149,7 @@ public class SocketFixedLengthEchoTest extends AbstractSocketTest {
             }
 
             if (channel.parent() != null) {
-                channel.write(msg);
+                channel.write(msg.retain());
             }
 
             counter += actual.length;
diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java b/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java
index 227a2d2791..c40c1d0b2d 100755
--- a/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java
+++ b/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java
@@ -1413,7 +1413,11 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
             outboundMessageBuffer().add(message);
         } else {
             ByteBuf buf = (ByteBuf) message;
-            outboundByteBuffer().writeBytes(buf, buf.readerIndex(), buf.readableBytes());
+            try {
+                outboundByteBuffer().writeBytes(buf, buf.readerIndex(), buf.readableBytes());
+            } finally {
+                buf.release();
+            }
         }
         invokeFlush0(promise);
     }