sync->sendAll/receiveAll

Author: technige

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

@@ -90,31 +90,50 @@ public void start()
         }
     }
 
-    public void sendAll( Queue<Message> messages ) throws IOException
+    public int sendAll( Queue<Message> messages ) throws IOException
     {
+        int messageCount = 0;
         while ( true )
         {
             Message message = messages.poll();
-            if ( message == null ) { break; }
-            else { writer.write( message ); }
+            if ( message == null )
+            {
+                break;
+            }
+            else
+            {
+                writer.write( message );
+                messageCount += 1;
+            }
+        }
+        if ( messageCount > 0 )
+        {
+            writer.flush();
         }
-        writer.flush();
+        return messageCount;
     }
 
-    public void receiveAll( SocketResponseHandler handler ) throws IOException
+    public int receiveAll( SocketResponseHandler handler ) throws IOException
     {
+        int messageCount = handler.collectorsWaiting();
         // Wait until all pending requests have been replied to
         while ( handler.collectorsWaiting() > 0 )
         {
-            reader.read( handler );
+            receiveOne( handler );
+        }
+        return messageCount;
+    }
 
-            // TODO: all the errors come from the following trace should result in the termination of this channel
-            // https://github.com/neo4j/neo4j/blob/3.0/community/bolt/src/main/java/org/neo4j/bolt/v1/transport/BoltProtocolV1.java#L86
-            if ( handler.protocolViolationErrorOccurred() )
-            {
-                stop();
-                throw handler.serverFailure();
-            }
+    public void receiveOne( SocketResponseHandler handler ) throws IOException
+    {
+        reader.read( handler );
+
+        // TODO: all the errors come from the following trace should result in the termination of this channel
+        // https://github.com/neo4j/neo4j/blob/3.0/community/bolt/src/main/java/org/neo4j/bolt/v1/transport/BoltProtocolV1.java#L86
+        if ( handler.protocolViolationErrorOccurred() )
+        {
+            stop();
+            throw handler.serverFailure();
         }
     }
 

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

@@ -24,10 +24,10 @@
 import java.util.Map;
 import java.util.Queue;
 
-import org.neo4j.driver.internal.messaging.ResetMessage;
 import org.neo4j.driver.internal.messaging.InitMessage;
 import org.neo4j.driver.internal.messaging.Message;
 import org.neo4j.driver.internal.messaging.PullAllMessage;
+import org.neo4j.driver.internal.messaging.ResetMessage;
 import org.neo4j.driver.internal.messaging.RunMessage;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.spi.Logger;
@@ -99,22 +99,40 @@ public void reset( StreamCollector collector )
     @Override
     public void sync()
     {
-        if ( pendingMessages.size() == 0 )
+        if ( sendAll() > 0 )
+        {
+            receiveAll();
+        }
+    }
+
+    @Override
+    public int sendAll()
+    {
+        try
+        {
+            return socket.sendAll( pendingMessages );
+        }
+        catch ( IOException e )
         {
-            return;
+            String message = e.getMessage();
+            throw new ClientException( "Unable to send messages to server: " + message, e );
         }
+    }
 
+    @Override
+    public int receiveAll()
+    {
         try
         {
-            socket.sendAll( pendingMessages );
-            socket.receiveAll( responseHandler );
+            int messageCount = socket.receiveAll( responseHandler );
             if ( responseHandler.serverFailureOccurred() )
             {
                 reset( StreamCollector.NO_OP );
                 Neo4jException exception = responseHandler.serverFailure();
                 responseHandler.reset();
                 throw exception;
             }
+            return messageCount;
         }
         catch ( IOException e )
         {

driver/src/main/java/org/neo4j/driver/internal/pool/PooledConnection.java

@@ -119,6 +119,34 @@ public void sync()
         }
     }
 
+    @Override
+    public int sendAll()
+    {
+        try
+        {
+            return delegate.sendAll();
+        }
+        catch ( RuntimeException e )
+        {
+            onDelegateException( e );
+            return -1;
+        }
+    }
+
+    @Override
+    public int receiveAll()
+    {
+        try
+        {
+            return delegate.receiveAll();
+        }
+        catch ( RuntimeException e )
+        {
+            onDelegateException( e );
+            return -1;
+        }
+    }
+
     @Override
     public void close()
     {

driver/src/main/java/org/neo4j/driver/internal/spi/Connection.java

@@ -63,6 +63,16 @@ public interface Connection extends AutoCloseable
      */
     void sync();
 
+    /**
+     * Send all pending messages to the server and return the number of messages sent.
+     */
+    int sendAll();
+
+    /**
+     * Receive all expected responses for any previous messages sent and return the number received.
+     */
+    int receiveAll();
+
     @Override
     void close();