Merge branch '1.0' into 1.1

Author: Zhen

driver/src/main/java/org/neo4j/driver/internal/NetworkSession.java

@@ -150,38 +150,47 @@ public void close()
         {
             throw new ClientException( "This session has already been closed." );
         }
-        else
+
+        if ( !connection.isOpen() )
         {
-            synchronized ( this )
+            // the socket connection is already closed due to some error, cannot send more data
+            closeConnection();
+            return;
+        }
+
+        synchronized ( this )
+        {
+            if ( currentTransaction != null )
             {
-                if ( currentTransaction != null )
+                try
                 {
-                    try
-                    {
-                        currentTransaction.close();
-                    }
-                    catch ( Throwable e )
-                    {
-                        // Best-effort
-                        logger.error( "Failed to close tx due to error: " + e.toString(), e );
-                    }
+                    currentTransaction.close();
+                }
+                catch ( Throwable e )
+                {
+                    // Best-effort
+                    logger.warn( "WARNING: Failed to close tx due to error: " + e.toString() );
                 }
             }
-            try
-            {
-                connection.sync();
-            }
-            catch( Throwable t )
-            {
-                logger.error( "Failed to sync messages due to error: " + t.toString(), t );
-                throw t;
-            }
-            finally
-            {
-                logger.debug( "~~ connection released by [session-%s]", sessionId );
-                connection.close();
-            }
         }
+        try
+        {
+            connection.sync();
+        }
+        catch ( Throwable t )
+        {
+            throw t;
+        }
+        finally
+        {
+            closeConnection();
+        }
+    }
+
+    private void closeConnection()
+    {
+        logger.debug( "~~ connection released by [session-%s]", sessionId );
+        connection.close();
     }
 
     @Override

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

@@ -0,0 +1,83 @@
+/**
+ * Copyright (c) 2002-2016 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * 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
+ *
+ * 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 org.neo4j.driver.internal.connector.socket;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ByteChannel;
+
+import org.neo4j.driver.internal.util.BytePrinter;
+import org.neo4j.driver.v1.exceptions.ClientException;
+
+/**
+ * Utility class for common operations.
+ */
+public final class SocketUtils
+{
+    private SocketUtils()
+    {
+        throw new UnsupportedOperationException( "Do not instantiate" );
+    }
+
+    public static void blockingRead(ByteChannel channel, ByteBuffer buf) throws IOException
+    {
+        while(buf.hasRemaining())
+        {
+            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 " +
+                        "instabilities, or due to restarts of the database. Expected %s bytes, received %s.",
+                        buf.limit(), bufStr.isEmpty() ? "none" : bufStr ) );
+            }
+        }
+    }
+
+    public static void blockingWrite(ByteChannel channel, ByteBuffer buf) throws IOException
+    {
+        while(buf.hasRemaining())
+        {
+            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 " +
+                        "instabilities, or due to restarts of the database. Expected %s bytes, wrote %s.",
+                        buf.limit(), bufStr.isEmpty() ? "none" :bufStr ) );
+            }
+        }
+    }
+}

driver/src/main/java/org/neo4j/driver/internal/net/BufferingChunkedInput.java

@@ -399,7 +399,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();
 
@@ -409,11 +409,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 ConnectionFailureException(
                         "Connection terminated while receiving data. This can happen due to network " +
-                        "instabilities, or due to restarts of the database.");
+                        "instabilities, or due to restarts of the database." );
             }
-            buffer.flip();
         }
         catch ( ClosedByInterruptException e )
         {
@@ -430,6 +437,10 @@ private static void readNextPacket( ReadableByteChannel channel, ByteBuffer buff
             throw new ConnectionFailureException(
                     "Unable to process request: " + message + " buffer: \n" + BytePrinter.hex( buffer ), e );
         }
+        finally
+        {
+            buffer.flip();
+        }
     }
 
     /**

driver/src/main/java/org/neo4j/driver/internal/net/SocketClient.java

@@ -76,6 +76,14 @@ void blockingRead( ByteBuffer buf ) throws IOException
         {
             if (channel.read( buf ) < 0)
             {
+                try
+                {
+                    channel.close();
+                }
+                catch ( IOException e )
+                {
+                    // best effort
+                }
                 String bufStr = BytePrinter.hex( buf ).trim();
                 throw new ConnectionFailureException( format(
                         "Connection terminated while receiving data. This can happen due to network " +
@@ -91,6 +99,14 @@ void blockingWrite( ByteBuffer buf ) throws IOException
         {
             if (channel.write( buf ) < 0)
             {
+                try
+                {
+                    channel.close();
+                }
+                catch ( IOException e )
+                {
+                    // best effort
+                }
                 String bufStr = BytePrinter.hex( buf ).trim();
                 throw new ConnectionFailureException( format(
                         "Connection terminated while sending data. This can happen due to network " +

driver/src/test/java/org/neo4j/driver/internal/net/BufferingChunkedInputTest.java

@@ -37,6 +37,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;
@@ -492,6 +493,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( ConnectionFailureException.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];
@@ -542,4 +565,4 @@ public void close() throws IOException
         };
     }
 
-}
+}