Better handling of closed connections

- Get rid of StackOverFlowError when no data on network
- If connection is closed give a proper error message

Author: pontusmelke

driver/src/main/java/org/neo4j/driver/internal/connector/socket/BufferingChunkedInput.java

@@ -1,15 +1,15 @@
 /**
  * Copyright (c) 2002-2016 "Neo Technology,"
  * Network Engine for Objects in Lund AB [http://neotechnology.com]
- *
+ * <p>
  * This file is part of Neo4j.
- *
+ * <p>
  * Licensed 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
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
  * 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.
@@ -100,8 +100,11 @@ public State readChunkSize( BufferingChunkedInput ctx ) throws IOException
                     {
                         if ( ctx.buffer.remaining() == 0 )
                         {
-                            //buffer empty, read next packet and try again
-                            readNextPacket( ctx.channel, ctx.buffer );
+                            //buffer empty, block until you get at least at least one byte
+                            while ( ctx.buffer.remaining() == 0 )
+                            {
+                                readNextPacket( ctx.channel, ctx.buffer );
+                            }
                             return AWAITING_CHUNK.readChunkSize( ctx );
                         }
                         else if ( ctx.buffer.remaining() >= 2 )
@@ -132,7 +135,7 @@ public State read( BufferingChunkedInput ctx ) throws IOException
                     public State peekByte( BufferingChunkedInput ctx ) throws IOException
                     {
                         //read chunk size and then proceed to read the rest of the chunk.
-                       return readChunkSize( ctx ).peekByte( ctx );
+                        return readChunkSize( ctx ).peekByte( ctx );
                     }
                 },
         IN_CHUNK
@@ -175,7 +178,7 @@ else if ( ctx.buffer.remaining() < ctx.scratchBuffer.remaining() )
                             int bytesToRead = min( ctx.scratchBuffer.remaining(), ctx.remainingChunkSize );
                             copyBytes( ctx.buffer, ctx.scratchBuffer, bytesToRead );
                             ctx.remainingChunkSize -= bytesToRead;
-                            if (ctx.scratchBuffer.remaining() == 0)
+                            if ( ctx.scratchBuffer.remaining() == 0 )
                             {
                                 //we have written all data that was asked for us
                                 return IN_CHUNK;
@@ -229,7 +232,7 @@ public State readChunkSize( BufferingChunkedInput ctx ) throws IOException
                     }
 
                     @Override
-                    public State read( BufferingChunkedInput ctx) throws IOException
+                    public State read( BufferingChunkedInput ctx ) throws IOException
                     {
                         throw new IllegalStateException( "Cannot read data while in progress of reading header" );
                     }
@@ -271,25 +274,34 @@ public State peekByte( BufferingChunkedInput ctx ) throws IOException
          * @param buffer The buffer to read into
          * @throws IOException
          */
-        private static void readNextPacket(ReadableByteChannel channel, ByteBuffer buffer ) throws IOException
+        private static void readNextPacket( ReadableByteChannel channel, ByteBuffer buffer ) throws IOException
         {
             try
             {
                 buffer.clear();
-                channel.read( buffer );
+                int read = channel.read( buffer );
+                if ( read == -1 )
+                {
+                    throw new ClientException(
+                            "Connection terminated while receiving data. This can happen due to network " +
+                            "instabilities, or due to restarts of the database." );
+                }
                 buffer.flip();
             }
-            catch( ClosedByInterruptException e )
+            catch ( ClosedByInterruptException e )
             {
                 throw new ClientException(
-                        "Connection to the database was lost because someone called `interrupt()` on the driver thread waiting for a reply. " +
-                        "This normally happens because the JVM is shutting down, but it can also happen because your application code or some " +
+                        "Connection to the database was lost because someone called `interrupt()` on the driver " +
+                        "thread waiting for a reply. " +
+                        "This normally happens because the JVM is shutting down, but it can also happen because your " +
+                        "application code or some " +
                         "framework you are using is manually interrupting the thread." );
             }
             catch ( IOException e )
             {
                 String message = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();
-                throw new ClientException( "Unable to process request: " + message + " buffer: \n" + BytePrinter.hex( buffer ), e );
+                throw new ClientException(
+                        "Unable to process request: " + message + " buffer: \n" + BytePrinter.hex( buffer ), e );
             }
         }
 
@@ -368,7 +380,7 @@ public PackInput readBytes( byte[] into, int offset, int toRead ) throws IOExcep
     public byte peekByte() throws IOException
     {
         state = state.peekByte( this );
-        return buffer.get(buffer.position());
+        return buffer.get( buffer.position() );
     }
 
     private boolean hasMoreDataUnreadInCurrentChunk()
@@ -382,7 +394,7 @@ private boolean hasMoreDataUnreadInCurrentChunk()
         public void run()
         {
             // the on message complete should only be called when no data unread from the message buffer
-            if( hasMoreDataUnreadInCurrentChunk() )
+            if ( hasMoreDataUnreadInCurrentChunk() )
             {
                 throw new ClientException( "Trying to read message complete ending '00 00' while there are more data " +
                                            "left in the message content unread: buffer [" +
@@ -392,11 +404,12 @@ public void run()
             try
             {
                 // read message boundary
-               state.readChunkSize( BufferingChunkedInput.this );
+                state.readChunkSize( BufferingChunkedInput.this );
                 if ( remainingChunkSize != 0 )
                 {
                     throw new ClientException( "Expecting message complete ending '00 00', but got " +
-                                               BytePrinter.hex( ByteBuffer.allocate( 2 ).putShort( (short) remainingChunkSize ) ) );
+                                               BytePrinter.hex( ByteBuffer.allocate( 2 )
+                                                       .putShort( (short) remainingChunkSize ) ) );
                 }
             }
             catch ( IOException e )

driver/src/test/java/org/neo4j/driver/internal/connector/socket/BufferingChunkedInputTest.java

@@ -1,15 +1,15 @@
 /**
  * Copyright (c) 2002-2016 "Neo Technology,"
  * Network Engine for Objects in Lund AB [http://neotechnology.com]
- *
+ * <p>
  * This file is part of Neo4j.
- *
+ * <p>
  * Licensed 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
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
  * 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.
@@ -21,7 +21,6 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
-import org.mockito.Matchers;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
@@ -38,6 +37,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -80,7 +80,8 @@ public void shouldReadOneByteInTwoChunks() throws IOException
     public void shouldReadOneByteWhenSplitHeader() throws IOException
     {
         // Given
-        BufferingChunkedInput input = new BufferingChunkedInput( packets( packet( 0 ), packet( 1, 13, 0, 1, 37, 0, 0 ) ) );
+        BufferingChunkedInput input =
+                new BufferingChunkedInput( packets( packet( 0 ), packet( 1, 13, 0, 1, 37, 0, 0 ) ) );
 
         // When
         byte b1 = input.readByte();
@@ -281,7 +282,7 @@ public void shouldGiveHelpfulMessageOnInterrupt() throws IOException
     {
         // Given
         ReadableByteChannel channel = mock( ReadableByteChannel.class );
-        when( channel.read( Matchers.any( ByteBuffer.class ) ) ).thenThrow( new ClosedByInterruptException() );
+        when( channel.read( any( ByteBuffer.class ) ) ).thenThrow( new ClosedByInterruptException() );
 
         BufferingChunkedInput ch = new BufferingChunkedInput( channel, 2 );
 
@@ -339,7 +340,8 @@ public void shouldPeekOneByteInTwoChunks() throws IOException
     public void shouldPeekOneByteWhenSplitHeader() throws IOException
     {
         // Given
-        BufferingChunkedInput input = new BufferingChunkedInput( packets( packet( 0 ), packet( 1, 13, 0, 1, 37, 0, 0 ) ) );
+        BufferingChunkedInput input =
+                new BufferingChunkedInput( packets( packet( 0 ), packet( 1, 13, 0, 1, 37, 0, 0 ) ) );
 
         // When
         byte peeked1 = input.peekByte();
@@ -373,6 +375,66 @@ public void shouldPeekOneByteInOneChunkWhenBustingBuffer() throws IOException
         assertThat( read2, equalTo( (byte) 37 ) );
     }
 
+    @Test
+    public void shouldNotStackOverflowWhenDataIsNotAvailable() throws IOException
+    {
+        // Given a channel that does not get data from the channel
+        ReadableByteChannel channel = new ReadableByteChannel()
+        {
+            private int counter = 0;
+            private int numberOfTries = 10000;
+
+            @Override
+            public int read( ByteBuffer dst ) throws IOException
+            {
+                if ( counter++ < numberOfTries )
+                {
+                    return 0;
+                }
+                else
+                {
+                    dst.put( (byte) 11 );
+                    return 1;
+                }
+            }
+
+            @Override
+            public boolean isOpen()
+            {
+                return true;
+            }
+
+            @Override
+            public void close() throws IOException
+            {
+
+            }
+        };
+
+        // When
+        BufferingChunkedInput input = new BufferingChunkedInput( channel );
+
+        // Then
+        assertThat(input.readByte(), equalTo( (byte)11 ));
+
+    }
+
+    @Test
+    public void shouldFailNicelyOnClosedConnections() throws IOException
+    {
+        // Given
+        ReadableByteChannel channel = mock( ReadableByteChannel.class );
+        when( channel.read( any( ByteBuffer.class ) ) ).thenReturn( -1 );
+        BufferingChunkedInput input = new BufferingChunkedInput( channel );
+
+        //Expect
+        exception.expect( ClientException.class );
+        exception.expectMessage( "Connection terminated while receiving data. This can happen due to network " +
+                                 "instabilities, or due to restarts of the database." );
+        // When
+        input.readByte();
+    }
+
     private ReadableByteChannel packet( int... bytes )
     {
         byte[] byteArray = new byte[bytes.length];