Better handling of closed connections

pontusmelke · pontusmelke · commit 49e3d73c3846 · 2016-04-15T12:29:18.000+02:00
- Get rid of StackOverFlowError when no data on network
- If connection is closed give a proper error message
diff --git a/driver/src/main/java/org/neo4j/driver/internal/connector/socket/BufferingChunkedInput.java b/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 [https://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
- *
- *     https://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
+ * https://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 )
diff --git a/driver/src/test/java/org/neo4j/driver/internal/connector/socket/BufferingChunkedInputTest.java b/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 [https://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
- *
- *     https://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
+ * https://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];

Original file line number	Diff line number	Diff line change
`@@ -1,15 +1,15 @@`
`1`	`1`	`/**`
`2`	`2`	`* Copyright (c) 2002-2016 "Neo Technology,"`
`3`	`3`	`* Network Engine for Objects in Lund AB [https://neotechnology.com]`
`4`		`- *`
	`4`	`+ * <p>`
`5`	`5`	`* This file is part of Neo4j.`
`6`		`- *`
	`6`	`+ * <p>`
`7`	`7`	`* Licensed under the Apache License, Version 2.0 (the "License");`
`8`	`8`	`* you may not use this file except in compliance with the License.`
`9`	`9`	`* You may obtain a copy of the License at`
`10`		`- *`
`11`		`- * https://www.apache.org/licenses/LICENSE-2.0`
`12`		`- *`
	`10`	`+ * <p>`
	`11`	`+ * https://www.apache.org/licenses/LICENSE-2.0`
	`12`	`+ * <p>`
`13`	`13`	`* Unless required by applicable law or agreed to in writing, software`
`14`	`14`	`* distributed under the License is distributed on an "AS IS" BASIS,`
`15`	`15`	`* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.`
`@@ -100,8 +100,11 @@ public State readChunkSize( BufferingChunkedInput ctx ) throws IOException`
`100`	`100`	`{`
`101`	`101`	`if ( ctx.buffer.remaining() == 0 )`
`102`	`102`	`{`
`103`		`- //buffer empty, read next packet and try again`
`104`		`- readNextPacket( ctx.channel, ctx.buffer );`
	`103`	`+ //buffer empty, block until you get at least at least one byte`
	`104`	`+ while ( ctx.buffer.remaining() == 0 )`
	`105`	`+ {`
	`106`	`+ readNextPacket( ctx.channel, ctx.buffer );`
	`107`	`+ }`
`105`	`108`	`return AWAITING_CHUNK.readChunkSize( ctx );`
`106`	`109`	`}`
`107`	`110`	`else if ( ctx.buffer.remaining() >= 2 )`
`@@ -132,7 +135,7 @@ public State read( BufferingChunkedInput ctx ) throws IOException`
`132`	`135`	`public State peekByte( BufferingChunkedInput ctx ) throws IOException`
`133`	`136`	`{`
`134`	`137`	`//read chunk size and then proceed to read the rest of the chunk.`
`135`		`- return readChunkSize( ctx ).peekByte( ctx );`
	`138`	`+ return readChunkSize( ctx ).peekByte( ctx );`
`136`	`139`	`}`
`137`	`140`	`},`
`138`	`141`	`IN_CHUNK`
`@@ -175,7 +178,7 @@ else if ( ctx.buffer.remaining() < ctx.scratchBuffer.remaining() )`
`175`	`178`	`int bytesToRead = min( ctx.scratchBuffer.remaining(), ctx.remainingChunkSize );`
`176`	`179`	`copyBytes( ctx.buffer, ctx.scratchBuffer, bytesToRead );`
`177`	`180`	`ctx.remainingChunkSize -= bytesToRead;`
`178`		`- if (ctx.scratchBuffer.remaining() == 0)`
	`181`	`+ if ( ctx.scratchBuffer.remaining() == 0 )`
`179`	`182`	`{`
`180`	`183`	`//we have written all data that was asked for us`
`181`	`184`	`return IN_CHUNK;`
`@@ -229,7 +232,7 @@ public State readChunkSize( BufferingChunkedInput ctx ) throws IOException`
`229`	`232`	`}`
`230`	`233`
`231`	`234`	`@Override`
`232`		`- public State read( BufferingChunkedInput ctx) throws IOException`
	`235`	`+ public State read( BufferingChunkedInput ctx ) throws IOException`
`233`	`236`	`{`
`234`	`237`	`throw new IllegalStateException( "Cannot read data while in progress of reading header" );`
`235`	`238`	`}`
`@@ -271,25 +274,34 @@ public State peekByte( BufferingChunkedInput ctx ) throws IOException`
`271`	`274`	`* @param buffer The buffer to read into`
`272`	`275`	`* @throws IOException`
`273`	`276`	`*/`
`274`		`- private static void readNextPacket(ReadableByteChannel channel, ByteBuffer buffer ) throws IOException`
	`277`	`+ private static void readNextPacket( ReadableByteChannel channel, ByteBuffer buffer ) throws IOException`
`275`	`278`	`{`
`276`	`279`	`try`
`277`	`280`	`{`
`278`	`281`	`buffer.clear();`
`279`		`- channel.read( buffer );`
	`282`	`+ int read = channel.read( buffer );`
	`283`	`+ if ( read == -1 )`
	`284`	`+ {`
	`285`	`+ throw new ClientException(`
	`286`	`+ "Connection terminated while receiving data. This can happen due to network " +`
	`287`	`+ "instabilities, or due to restarts of the database." );`
	`288`	`+ }`
`280`	`289`	`buffer.flip();`
`281`	`290`	`}`
`282`		`- catch( ClosedByInterruptException e )`
	`291`	`+ catch ( ClosedByInterruptException e )`
`283`	`292`	`{`
`284`	`293`	`throw new ClientException(`
`285`		- "Connection to the database was lost because someone called `interrupt()` on the driver thread waiting for a reply. " +
`286`		`- "This normally happens because the JVM is shutting down, but it can also happen because your application code or some " +`
	`294`	+ "Connection to the database was lost because someone called `interrupt()` on the driver " +
	`295`	`+ "thread waiting for a reply. " +`
	`296`	`+ "This normally happens because the JVM is shutting down, but it can also happen because your " +`
	`297`	`+ "application code or some " +`
`287`	`298`	`"framework you are using is manually interrupting the thread." );`
`288`	`299`	`}`
`289`	`300`	`catch ( IOException e )`
`290`	`301`	`{`
`291`	`302`	`String message = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();`
`292`		`- throw new ClientException( "Unable to process request: " + message + " buffer: \n" + BytePrinter.hex( buffer ), e );`
	`303`	`+ throw new ClientException(`
	`304`	`+ "Unable to process request: " + message + " buffer: \n" + BytePrinter.hex( buffer ), e );`
`293`	`305`	`}`
`294`	`306`	`}`
`295`	`307`
`@@ -368,7 +380,7 @@ public PackInput readBytes( byte[] into, int offset, int toRead ) throws IOExcep`
`368`	`380`	`public byte peekByte() throws IOException`
`369`	`381`	`{`
`370`	`382`	`state = state.peekByte( this );`
`371`		`- return buffer.get(buffer.position());`
	`383`	`+ return buffer.get( buffer.position() );`
`372`	`384`	`}`
`373`	`385`
`374`	`386`	`private boolean hasMoreDataUnreadInCurrentChunk()`
`@@ -382,7 +394,7 @@ private boolean hasMoreDataUnreadInCurrentChunk()`
`382`	`394`	`public void run()`
`383`	`395`	`{`
`384`	`396`	`// the on message complete should only be called when no data unread from the message buffer`
`385`		`- if( hasMoreDataUnreadInCurrentChunk() )`
	`397`	`+ if ( hasMoreDataUnreadInCurrentChunk() )`
`386`	`398`	`{`
`387`	`399`	`throw new ClientException( "Trying to read message complete ending '00 00' while there are more data " +`
`388`	`400`	`"left in the message content unread: buffer [" +`
`@@ -392,11 +404,12 @@ public void run()`
`392`	`404`	`try`
`393`	`405`	`{`
`394`	`406`	`// read message boundary`
`395`		`- state.readChunkSize( BufferingChunkedInput.this );`
	`407`	`+ state.readChunkSize( BufferingChunkedInput.this );`
`396`	`408`	`if ( remainingChunkSize != 0 )`
`397`	`409`	`{`
`398`	`410`	`throw new ClientException( "Expecting message complete ending '00 00', but got " +`
`399`		`- BytePrinter.hex( ByteBuffer.allocate( 2 ).putShort( (short) remainingChunkSize ) ) );`
	`411`	`+ BytePrinter.hex( ByteBuffer.allocate( 2 )`
	`412`	`+ .putShort( (short) remainingChunkSize ) ) );`
`400`	`413`	`}`
`401`	`414`	`}`
`402`	`415`	`catch ( IOException e )`