Merge pull request #429 from lutovich/1.5-buffer-handling

Fixed couple netty buffer leaks

Author: lutovich

driver/src/main/java/org/neo4j/driver/internal/async/inbound/MessageDecoder.java

@@ -41,21 +41,21 @@ public void channelRead( ChannelHandlerContext ctx, Object msg ) throws Exceptio
     }
 
     @Override
-    protected void decode( ChannelHandlerContext ctx, ByteBuf in, List<Object> out ) throws Exception
+    protected void decode( ChannelHandlerContext ctx, ByteBuf in, List<Object> out )
     {
         if ( readMessageBoundary )
         {
             // now we have a complete message in the input buffer
 
-            // increment ref count of the buffer because we will pass it's duplicate through
-            in.retain();
-            ByteBuf res = in.duplicate();
+            // increment ref count of the buffer and create it's duplicate that shares the content
+            // duplicate will be the output of this decoded and input for the next one
+            ByteBuf messageBuf = in.retainedDuplicate();
 
             // signal that whole message was read by making input buffer seem like it was fully read/consumed
             in.readerIndex( in.readableBytes() );
 
             // pass the full message to the next handler in the pipeline
-            out.add( res );
+            out.add( messageBuf );
 
             readMessageBoundary = false;
         }

driver/src/main/java/org/neo4j/driver/internal/async/outbound/OutboundMessageHandler.java

@@ -65,14 +65,14 @@ protected void encode( ChannelHandlerContext ctx, Message msg, List<Object> out
         try
         {
             writer.write( msg );
+            output.stop();
         }
         catch ( Throwable error )
-        {
-            throw new EncoderException( "Failed to write outbound message: " + msg, error );
-        }
-        finally
         {
             output.stop();
+            // release buffer because it will not get added to the out list and no other handler is going to handle it
+            messageBuf.release();
+            throw new EncoderException( "Failed to write outbound message: " + msg, error );
         }
 
         if ( log.isTraceEnabled() )

driver/src/test/java/org/neo4j/driver/internal/async/ChannelAttributesTest.java

@@ -19,7 +19,6 @@
 package org.neo4j.driver.internal.async;
 
 import io.netty.channel.embedded.EmbeddedChannel;
-import org.junit.After;
 import org.junit.Test;
 
 import org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher;
@@ -47,12 +46,6 @@ public class ChannelAttributesTest
 {
     private final EmbeddedChannel channel = new EmbeddedChannel();
 
-    @After
-    public void tearDown() throws Exception
-    {
-        channel.close();
-    }
-
     @Test
     public void shouldSetAndGetAddress()
     {

driver/src/test/java/org/neo4j/driver/internal/async/ChannelConnectedListenerTest.java

@@ -45,7 +45,7 @@ public class ChannelConnectedListenerTest
     @After
     public void tearDown()
     {
-        channel.close();
+        channel.finishAndReleaseAll();
     }
 
     @Test

driver/src/test/java/org/neo4j/driver/internal/async/HandshakeCompletedListenerTest.java

@@ -49,9 +49,9 @@ public class HandshakeCompletedListenerTest
     private final EmbeddedChannel channel = new EmbeddedChannel();
 
     @After
-    public void tearDown() throws Exception
+    public void tearDown()
     {
-        channel.close();
+        channel.finishAndReleaseAll();
     }
 
     @Test

driver/src/test/java/org/neo4j/driver/internal/async/HandshakeResponseHandlerTest.java

@@ -59,7 +59,7 @@ public void setUp()
     @After
     public void tearDown()
     {
-        channel.close();
+        channel.finishAndReleaseAll();
     }
 
     @Test

driver/src/test/java/org/neo4j/driver/internal/async/NettyChannelInitializerTest.java

@@ -20,6 +20,7 @@
 
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.ssl.SslHandler;
+import org.junit.After;
 import org.junit.Test;
 
 import org.neo4j.driver.internal.security.SecurityPlan;
@@ -31,20 +32,27 @@
 import static org.junit.Assert.assertNull;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
+import static org.neo4j.driver.internal.async.BoltServerAddress.LOCAL_DEFAULT;
 import static org.neo4j.driver.internal.async.ChannelAttributes.creationTimestamp;
 import static org.neo4j.driver.internal.async.ChannelAttributes.messageDispatcher;
 import static org.neo4j.driver.internal.async.ChannelAttributes.serverAddress;
-import static org.neo4j.driver.internal.async.BoltServerAddress.LOCAL_DEFAULT;
 
 public class NettyChannelInitializerTest
 {
+    private final EmbeddedChannel channel = new EmbeddedChannel();
+
+    @After
+    public void tearDown()
+    {
+        channel.finishAndReleaseAll();
+    }
+
     @Test
     public void shouldAddSslHandlerWhenRequiresEncryption() throws Exception
     {
         SecurityPlan security = SecurityPlan.forAllCertificates();
         NettyChannelInitializer initializer = new NettyChannelInitializer( LOCAL_DEFAULT, security, new FakeClock() );
 
-        EmbeddedChannel channel = new EmbeddedChannel();
         initializer.initChannel( channel );
 
         assertNotNull( channel.pipeline().get( SslHandler.class ) );
@@ -56,7 +64,6 @@ public void shouldNotAddSslHandlerWhenDoesNotRequireEncryption()
         SecurityPlan security = SecurityPlan.insecure();
         NettyChannelInitializer initializer = new NettyChannelInitializer( LOCAL_DEFAULT, security, new FakeClock() );
 
-        EmbeddedChannel channel = new EmbeddedChannel();
         initializer.initChannel( channel );
 
         assertNull( channel.pipeline().get( SslHandler.class ) );
@@ -70,7 +77,6 @@ public void shouldUpdateChannelAttributes()
         SecurityPlan security = SecurityPlan.insecure();
         NettyChannelInitializer initializer = new NettyChannelInitializer( LOCAL_DEFAULT, security, clock );
 
-        EmbeddedChannel channel = new EmbeddedChannel();
         initializer.initChannel( channel );
 
         assertEquals( LOCAL_DEFAULT, serverAddress( channel ) );

driver/src/test/java/org/neo4j/driver/internal/async/inbound/ChunkDecoderTest.java

@@ -20,6 +20,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.After;
 import org.junit.Test;
 
 import static io.netty.buffer.Unpooled.buffer;
@@ -28,9 +29,18 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.neo4j.driver.v1.util.TestUtil.assertByteBufEquals;
 
 public class ChunkDecoderTest
 {
+    private final EmbeddedChannel channel = new EmbeddedChannel( new ChunkDecoder() );
+
+    @After
+    public void tearDown()
+    {
+        channel.finishAndReleaseAll();
+    }
+
     @Test
     public void shouldDecodeFullChunk()
     {
@@ -54,7 +64,7 @@ public void shouldDecodeFullChunk()
         // there should only be a single chunk available for reading
         assertEquals( 1, channel.inboundMessages().size() );
         // it should have no size header and expected body
-        assertEquals( input.slice( 2, 7 ), channel.readInbound() );
+        assertByteBufEquals( input.slice( 2, 7 ), channel.readInbound() );
     }
 
     @Test
@@ -97,7 +107,7 @@ public void shouldDecodeSplitChunk()
         // there should only be a single chunk available for reading
         assertEquals( 1, channel.inboundMessages().size() );
         // it should have no size header and expected body
-        assertEquals( wrappedBuffer( new byte[]{1, 11, 2, 22, 3, 33, 4, 44, 5} ), channel.readInbound() );
+        assertByteBufEquals( wrappedBuffer( new byte[]{1, 11, 2, 22, 3, 33, 4, 44, 5} ), channel.readInbound() );
     }
 
     @Test
@@ -113,6 +123,6 @@ public void shouldDecodeEmptyChunk()
         // there should only be a single chunk available for reading
         assertEquals( 1, channel.inboundMessages().size() );
         // it should have no size header and empty body
-        assertEquals( wrappedBuffer( new byte[0] ), channel.readInbound() );
+        assertByteBufEquals( wrappedBuffer( new byte[0] ), channel.readInbound() );
     }
 }

driver/src/test/java/org/neo4j/driver/internal/async/inbound/InboundMessageHandlerTest.java

@@ -76,7 +76,7 @@ public void tearDown()
     {
         if ( channel != null )
         {
-            channel.close();
+            channel.finishAndReleaseAll();
         }
     }
 

driver/src/test/java/org/neo4j/driver/internal/async/inbound/MessageDecoderTest.java

@@ -19,41 +19,47 @@
 package org.neo4j.driver.internal.async.inbound;
 
 import io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.After;
 import org.junit.Test;
 
 import static io.netty.buffer.Unpooled.wrappedBuffer;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.neo4j.driver.v1.util.TestUtil.assertByteBufEquals;
 
 public class MessageDecoderTest
 {
+    private final EmbeddedChannel channel = new EmbeddedChannel( new MessageDecoder() );
+
+    @After
+    public void tearDown()
+    {
+        channel.finishAndReleaseAll();
+    }
+
     @Test
     public void shouldDecodeMessageWithSingleChunk()
     {
-        EmbeddedChannel channel = new EmbeddedChannel( new MessageDecoder() );
-
         assertFalse( channel.writeInbound( wrappedBuffer( new byte[]{1, 2, 3, 4, 5} ) ) );
         assertTrue( channel.writeInbound( wrappedBuffer( new byte[0] ) ) );
         assertTrue( channel.finish() );
 
         assertEquals( 1, channel.inboundMessages().size() );
-        assertEquals( wrappedBuffer( new byte[]{1, 2, 3, 4, 5} ), channel.readInbound() );
+        assertByteBufEquals( wrappedBuffer( new byte[]{1, 2, 3, 4, 5} ), channel.readInbound() );
     }
 
     @Test
     public void shouldDecodeMessageWithMultipleChunks()
     {
-        EmbeddedChannel channel = new EmbeddedChannel( new MessageDecoder() );
-
         assertFalse( channel.writeInbound( wrappedBuffer( new byte[]{1, 2, 3} ) ) );
         assertFalse( channel.writeInbound( wrappedBuffer( new byte[]{4, 5} ) ) );
         assertFalse( channel.writeInbound( wrappedBuffer( new byte[]{6, 7, 8} ) ) );
         assertTrue( channel.writeInbound( wrappedBuffer( new byte[0] ) ) );
         assertTrue( channel.finish() );
 
         assertEquals( 1, channel.inboundMessages().size() );
-        assertEquals( wrappedBuffer( new byte[]{1, 2, 3, 4, 5, 6, 7, 8} ), channel.readInbound() );
+        assertByteBufEquals( wrappedBuffer( new byte[]{1, 2, 3, 4, 5, 6, 7, 8} ), channel.readInbound() );
     }
 
     @Test
@@ -73,8 +79,8 @@ public void shouldDecodeMultipleConsecutiveMessages()
         channel.writeInbound( wrappedBuffer( new byte[0] ) );
 
         assertEquals( 3, channel.inboundMessages().size() );
-        assertEquals( wrappedBuffer( new byte[]{1, 2, 3} ), channel.readInbound() );
-        assertEquals( wrappedBuffer( new byte[]{4, 5, 6} ), channel.readInbound() );
-        assertEquals( wrappedBuffer( new byte[]{7, 8, 9, 10} ), channel.readInbound() );
+        assertByteBufEquals( wrappedBuffer( new byte[]{1, 2, 3} ), channel.readInbound() );
+        assertByteBufEquals( wrappedBuffer( new byte[]{4, 5, 6} ), channel.readInbound() );
+        assertByteBufEquals( wrappedBuffer( new byte[]{7, 8, 9, 10} ), channel.readInbound() );
     }
 }