Merge pull request #150 from pontusmelke/1.0-chunking-state-machine

Buffer channel reads

Author: jakewins

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

@@ -228,7 +228,7 @@ public static ByteChannel create( String host, int port, Config config, Logger l
             }
             case NONE:
             {
-                channel = new AllOrNothingChannel( soChannel );
+                channel = soChannel;
                 break;
             }
             default:

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

@@ -37,7 +37,7 @@ public SocketProtocolV1( ByteChannel channel ) throws IOException
         messageFormat = new PackStreamMessageFormatV1();
 
         ChunkedOutput output = new ChunkedOutput( channel );
-        ChunkedInput input = new ChunkedInput( channel );
+        BufferingChunkedInput input = new BufferingChunkedInput( channel );
 
         this.writer = new PackStreamMessageFormatV1.Writer( output, output.messageBoundaryHook() );
         this.reader = new PackStreamMessageFormatV1.Reader( input, input.messageBoundaryHook() );

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

@@ -33,7 +33,9 @@
 import org.neo4j.driver.v1.exceptions.DatabaseException;
 import org.neo4j.driver.v1.exceptions.Neo4jException;
 import org.neo4j.driver.v1.exceptions.TransientException;
+import org.neo4j.driver.v1.summary.Notification;
 import org.neo4j.driver.v1.summary.StatementType;
+import org.neo4j.driver.v1.util.Function;
 
 public class SocketResponseHandler implements MessageHandler
 {
@@ -95,8 +97,9 @@ private void collectNotifications( StreamCollector collector, Value notification
     {
         if ( notifications != null )
         {
-            collector.notifications( notifications.asList( InternalNotification
-                    .VALUE_TO_NOTIFICATION ) );
+            Function<Value,Notification> notification = InternalNotification
+                    .VALUE_TO_NOTIFICATION;
+            collector.notifications( notifications.asList( notification ) );
         }
     }
 

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

@@ -64,6 +64,8 @@ public class TLSSocketChannel implements ByteChannel
     private ByteBuffer plainIn;
     private ByteBuffer plainOut;
 
+    private static final ByteBuffer DUMMY_BUFFER = ByteBuffer.allocateDirect( 0 );
+
     public TLSSocketChannel( String host, int port, SocketChannel channel, Logger logger,
                              TrustStrategy trustStrategy )
             throws GeneralSecurityException, IOException
@@ -123,7 +125,7 @@ private void runHandshake() throws IOException
                 break;
             case NEED_UNWRAP:
                 // Unwrap the ssl packet to value ssl handshake information
-                handshakeStatus = unwrap( null );
+                handshakeStatus = unwrap( DUMMY_BUFFER );
                 plainIn.clear();
                 break;
             case NEED_WRAP:
@@ -329,17 +331,17 @@ private HandshakeStatus wrap( ByteBuffer buffer ) throws IOException
      */
     static int bufferCopy( ByteBuffer from, ByteBuffer to )
     {
-        if ( from == null || to == null )
-        {
-            return 0;
-        }
+        int maxTransfer = Math.min( to.remaining(), from.remaining() );
 
-        int i;
-        for ( i = 0; to.remaining() > 0 && from.remaining() > 0; i++ )
-        {
-            to.put( from.get() );
-        }
-        return i;
+        //use a temp buffer and move all data in one go
+        ByteBuffer temporaryBuffer = from.duplicate();
+        temporaryBuffer.limit( temporaryBuffer.position() + maxTransfer );
+        to.put( temporaryBuffer );
+
+        //move postion so it appears as if we read the buffer
+        from.position( from.position() + maxTransfer );
+
+        return maxTransfer;
     }
 
     /**
@@ -391,23 +393,18 @@ public int read( ByteBuffer dst ) throws IOException
          */
         int toRead = dst.remaining();
         plainIn.flip();
-        if ( plainIn.remaining() >= toRead )
+        if ( plainIn.hasRemaining() )
         {
             bufferCopy( plainIn, dst );
             plainIn.compact();
         }
         else
         {
-            dst.put( plainIn );             // Copy whatever left in the plainIn to dst
-            do
-            {
-                plainIn.clear();            // Clear plainIn
-                unwrap( dst );              // Read more data from the underline channel and save the data read into dst
-            }
-            while ( dst.remaining() > 0 );  // If enough bytes read then return otherwise continue reading from channel
+            plainIn.clear();            // Clear plainIn
+            unwrap( dst );              // Read more data from the underline channel and save the data read into dst
         }
 
-        return toRead;
+        return toRead - dst.remaining();
     }
 
     @Override

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

@@ -31,7 +31,7 @@
 import org.neo4j.driver.internal.InternalNode;
 import org.neo4j.driver.internal.InternalPath;
 import org.neo4j.driver.internal.InternalRelationship;
-import org.neo4j.driver.internal.connector.socket.ChunkedInput;
+import org.neo4j.driver.internal.connector.socket.BufferingChunkedInput;
 import org.neo4j.driver.internal.connector.socket.ChunkedOutput;
 import org.neo4j.driver.internal.packstream.PackInput;
 import org.neo4j.driver.internal.packstream.PackOutput;
@@ -44,12 +44,12 @@
 import org.neo4j.driver.internal.value.NodeValue;
 import org.neo4j.driver.internal.value.PathValue;
 import org.neo4j.driver.internal.value.RelationshipValue;
+import org.neo4j.driver.v1.Value;
+import org.neo4j.driver.v1.exceptions.ClientException;
 import org.neo4j.driver.v1.types.Entity;
 import org.neo4j.driver.v1.types.Node;
 import org.neo4j.driver.v1.types.Path;
 import org.neo4j.driver.v1.types.Relationship;
-import org.neo4j.driver.v1.Value;
-import org.neo4j.driver.v1.exceptions.ClientException;
 
 import static org.neo4j.driver.v1.Values.value;
 
@@ -87,7 +87,7 @@ public MessageFormat.Writer newWriter( WritableByteChannel ch )
     @Override
     public MessageFormat.Reader newReader( ReadableByteChannel ch )
     {
-        ChunkedInput input = new ChunkedInput( ch );
+        BufferingChunkedInput input = new BufferingChunkedInput( ch );
         return new Reader( input, input.messageBoundaryHook() );
     }
 
@@ -534,7 +534,8 @@ private Value unpackValue() throws IOException
                 {
                 case NODE:
                     ensureCorrectStructSize( "NODE", NODE_FIELDS, size );
-                    return new NodeValue( unpackNode() );
+                    InternalNode adapted = unpackNode();
+                    return new NodeValue( adapted );
                 case RELATIONSHIP:
                     ensureCorrectStructSize( "RELATIONSHIP", 5, size );
                     return unpackRelationship();
@@ -555,7 +556,8 @@ private Value unpackRelationship() throws IOException
             String relType = unpacker.unpackString();
             Map<String,Value> props = unpackMap();
 
-            return new RelationshipValue( new InternalRelationship( urn, startUrn, endUrn, relType, props ) );
+            InternalRelationship adapted = new InternalRelationship( urn, startUrn, endUrn, relType, props );
+            return new RelationshipValue( adapted );
         }
 
         private InternalNode unpackNode() throws IOException

driver/src/main/java/org/neo4j/driver/internal/messaging/PackStreamMessageFormatV1.java

@@ -30,10 +30,11 @@ public class BufferedChannelInput implements PackInput
 {
     private final ByteBuffer buffer;
     private ReadableByteChannel channel;
+    private static final int DEFAULT_BUFFER_CAPACITY = 8192;
 
-    public BufferedChannelInput( int bufferCapacity )
+    public BufferedChannelInput(ReadableByteChannel ch )
     {
-        this( bufferCapacity, null );
+        this( DEFAULT_BUFFER_CAPACITY, ch );
     }
 
     public BufferedChannelInput( int bufferCapacity, ReadableByteChannel ch )

driver/src/main/java/org/neo4j/driver/internal/packstream/BufferedChannelInput.java

@@ -19,7 +19,6 @@
 package org.neo4j.driver.internal.packstream;
 
 import java.io.IOException;
-import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.WritableByteChannel;
 import java.nio.charset.Charset;
 import java.util.List;
@@ -147,8 +146,6 @@ public class PackStream
     private static final String EMPTY_STRING = "";
     private static final Charset UTF_8 = Charset.forName( "UTF-8" );
 
-    private static final int DEFAULT_BUFFER_CAPACITY = 8192;
-
     private PackStream() {}
 
     public static class Packer
@@ -428,29 +425,11 @@ public static class Unpacker
     {
         private PackInput in;
 
-        public Unpacker( ReadableByteChannel channel )
-        {
-            this( DEFAULT_BUFFER_CAPACITY );
-            reset( channel );
-        }
-
-        public Unpacker( int bufferCapacity )
-        {
-            assert bufferCapacity >= 8 : "Buffer must be at least 8 bytes.";
-            this.in = new BufferedChannelInput( bufferCapacity );
-        }
-
         public Unpacker( PackInput in )
         {
             this.in = in;
         }
 
-        public Unpacker reset( ReadableByteChannel ch )
-        {
-            ((BufferedChannelInput)in).reset( ch );
-            return this;
-        }
-
         public boolean hasNext() throws IOException
         {
             return in.hasMoreData();

driver/src/main/java/org/neo4j/driver/internal/packstream/PackStream.java

@@ -24,6 +24,7 @@
  */
 public class Functions
 {
+    @SuppressWarnings( "unchecked" )
     public static <T> Function<T,T> identity()
     {
         return IDENTITY;