Merge pull request #249 from zhenlineo/1.0-buffer-flip

Zhen Li · web-flow · commit 68e63448bed4 · 2016-10-14T10:55:32.000+02:00
Fixed expect strust but get xx bug
diff --git a/driver/src/main/java/org/neo4j/driver/internal/InternalSession.java b/driver/src/main/java/org/neo4j/driver/internal/InternalSession.java
@@ -108,32 +108,38 @@ public boolean isOpen()
     public void close()
     {
         // Use atomic operation to protect from closing the connection twice (putting back to the pool twice).
-        if( !isOpen.compareAndSet( true, false ) )
+        if ( !isOpen.compareAndSet( true, false ) )
         {
             throw new ClientException( "This session has already been closed." );
         }
-        else
+
+        if( !connection.isOpen() )
+        {
+            // the socket connection is already closed due to some error, cannot send more data
+            connection.close();
+            return;
+        }
+
+        if ( currentTransaction != null )
         {
-            if ( currentTransaction != null )
-            {
-                try
-                {
-                    currentTransaction.close();
-                }
-                catch ( Throwable e )
-                {
-                    // Best-effort
-                }
-            }
             try
             {
-                connection.sync();
+                currentTransaction.close();
             }
-            finally
+            catch ( Throwable e )
             {
-                connection.close();
+                // Best-effort
             }
         }
+        try
+        {
+            connection.sync();
+        }
+        finally
+        {
+            connection.close();
+        }
+
     }
 
     @Override
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
@@ -398,7 +398,7 @@ else if ( buffer.remaining() >= 2 )
      * @param buffer The buffer to read into
      * @throws IOException
      */
-    private static void readNextPacket( ReadableByteChannel channel, ByteBuffer buffer ) throws IOException
+    static void readNextPacket( ReadableByteChannel channel, ByteBuffer buffer ) throws IOException
     {
         assert !buffer.hasRemaining();
 
@@ -408,11 +408,18 @@ private static void readNextPacket( ReadableByteChannel channel, ByteBuffer buff
             int read = channel.read( buffer );
             if ( read == -1 )
             {
+                try
+                {
+                    channel.close();
+                }
+                catch ( IOException e )
+                {
+                    // best effort
+                }
                 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 )
         {
@@ -429,6 +436,10 @@ private static void readNextPacket( ReadableByteChannel channel, ByteBuffer buff
             throw new ClientException(
                     "Unable to process request: " + message + " buffer: \n" + BytePrinter.hex( buffer ), e );
         }
+        finally
+        {
+            buffer.flip();
+        }
     }
 
     /**
diff --git a/driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketUtils.java b/driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketUtils.java
@@ -41,6 +41,14 @@ public static void blockingRead(ByteChannel channel, ByteBuffer buf) throws IOEx
         {
             if (channel.read( buf ) < 0)
             {
+                try
+                {
+                    channel.close();
+                }
+                catch ( IOException e )
+                {
+                    // best effort
+                }
                 String bufStr = BytePrinter.hex( buf ).trim();
                 throw new ClientException( String.format(
                         "Connection terminated while receiving data. This can happen due to network " +
@@ -56,6 +64,14 @@ public static void blockingWrite(ByteChannel channel, ByteBuffer buf) throws IOE
         {
             if (channel.write( buf ) < 0)
             {
+                try
+                {
+                    channel.close();
+                }
+                catch ( IOException e )
+                {
+                    // best effort
+                }
                 String bufStr = BytePrinter.hex( buf ).trim();
                 throw new ClientException( String.format(
                         "Connection terminated while sending data. This can happen due to network " +
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
@@ -36,6 +36,7 @@
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
@@ -491,6 +492,28 @@ public void shouldFailNicelyOnClosedConnections() throws IOException
         input.readByte();
     }
 
+
+    @Test
+    public void shouldKeepBufferCorrectWhenError() throws Throwable
+    {
+        // Given
+        ReadableByteChannel channel = mock( ReadableByteChannel.class );
+        when( channel.read( any( ByteBuffer.class ) ) ).thenReturn( -1 );
+        ByteBuffer buffer = ByteBuffer.allocate( 8 );
+        buffer.limit(0);
+
+        //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
+        BufferingChunkedInput.readNextPacket( channel, buffer );
+        assertEquals( buffer.position(), 0 );
+        assertEquals( buffer.limit(), 0 );
+        assertEquals( buffer.capacity(), 8 );
+        assertFalse( channel.isOpen() );
+    }
+
     private ReadableByteChannel fillPacket( int size, int value )
     {
         int[] ints = new int[size];
@@ -541,4 +564,4 @@ public void close() throws IOException
         };
     }
 
-}
+}

Original file line number	Diff line number	Diff line change
`@@ -108,32 +108,38 @@ public boolean isOpen()`
`108`	`108`	`public void close()`
`109`	`109`	`{`
`110`	`110`	`// Use atomic operation to protect from closing the connection twice (putting back to the pool twice).`
`111`		`- if( !isOpen.compareAndSet( true, false ) )`
	`111`	`+ if ( !isOpen.compareAndSet( true, false ) )`
`112`	`112`	`{`
`113`	`113`	`throw new ClientException( "This session has already been closed." );`
`114`	`114`	`}`
`115`		`- else`
	`115`	`+`
	`116`	`+ if( !connection.isOpen() )`
	`117`	`+ {`
	`118`	`+ // the socket connection is already closed due to some error, cannot send more data`
	`119`	`+ connection.close();`
	`120`	`+ return;`
	`121`	`+ }`
	`122`	`+`
	`123`	`+ if ( currentTransaction != null )`
`116`	`124`	`{`
`117`		`- if ( currentTransaction != null )`
`118`		`- {`
`119`		`- try`
`120`		`- {`
`121`		`- currentTransaction.close();`
`122`		`- }`
`123`		`- catch ( Throwable e )`
`124`		`- {`
`125`		`- // Best-effort`
`126`		`- }`
`127`		`- }`
`128`	`125`	`try`
`129`	`126`	`{`
`130`		`- connection.sync();`
	`127`	`+ currentTransaction.close();`
`131`	`128`	`}`
`132`		`- finally`
	`129`	`+ catch ( Throwable e )`
`133`	`130`	`{`
`134`		`- connection.close();`
	`131`	`+ // Best-effort`
`135`	`132`	`}`
`136`	`133`	`}`
	`134`	`+ try`
	`135`	`+ {`
	`136`	`+ connection.sync();`
	`137`	`+ }`
	`138`	`+ finally`
	`139`	`+ {`
	`140`	`+ connection.close();`
	`141`	`+ }`
	`142`	`+`
`137`	`143`	`}`
`138`	`144`
`139`	`145`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -398,7 +398,7 @@ else if ( buffer.remaining() >= 2 )`
`398`	`398`	`* @param buffer The buffer to read into`
`399`	`399`	`* @throws IOException`
`400`	`400`	`*/`
`401`		`- private static void readNextPacket( ReadableByteChannel channel, ByteBuffer buffer ) throws IOException`
	`401`	`+ static void readNextPacket( ReadableByteChannel channel, ByteBuffer buffer ) throws IOException`
`402`	`402`	`{`
`403`	`403`	`assert !buffer.hasRemaining();`
`404`	`404`
`@@ -408,11 +408,18 @@ private static void readNextPacket( ReadableByteChannel channel, ByteBuffer buff`
`408`	`408`	`int read = channel.read( buffer );`
`409`	`409`	`if ( read == -1 )`
`410`	`410`	`{`
	`411`	`+ try`
	`412`	`+ {`
	`413`	`+ channel.close();`
	`414`	`+ }`
	`415`	`+ catch ( IOException e )`
	`416`	`+ {`
	`417`	`+ // best effort`
	`418`	`+ }`
`411`	`419`	`throw new ClientException(`
`412`	`420`	`"Connection terminated while receiving data. This can happen due to network " +`
`413`	`421`	`"instabilities, or due to restarts of the database." );`
`414`	`422`	`}`
`415`		`- buffer.flip();`
`416`	`423`	`}`
`417`	`424`	`catch ( ClosedByInterruptException e )`
`418`	`425`	`{`
`@@ -429,6 +436,10 @@ private static void readNextPacket( ReadableByteChannel channel, ByteBuffer buff`
`429`	`436`	`throw new ClientException(`
`430`	`437`	`"Unable to process request: " + message + " buffer: \n" + BytePrinter.hex( buffer ), e );`
`431`	`438`	`}`
	`439`	`+ finally`
	`440`	`+ {`
	`441`	`+ buffer.flip();`
	`442`	`+ }`
`432`	`443`	`}`
`433`	`444`
`434`	`445`	`/**`