Merge branch 1.6 into 1.7

Author: lutovich

driver/src/main/java/org/neo4j/driver/internal/async/DirectConnection.java

@@ -199,7 +199,7 @@ private void writeResetMessageIfNeeded( ResponseHandler resetHandler, boolean is
                 // auto-read could've been disabled, re-enable it to automatically receive response for RESET
                 setAutoRead( true );
 
-                messageDispatcher.queue( resetHandler );
+                messageDispatcher.enqueue( resetHandler );
                 channel.writeAndFlush( ResetMessage.RESET, channel.voidPromise() );
             }
         } );
@@ -209,7 +209,7 @@ private void writeMessageInEventLoop( Message message, ResponseHandler handler,
     {
         channel.eventLoop().execute( () ->
         {
-            messageDispatcher.queue( handler );
+            messageDispatcher.enqueue( handler );
 
             if ( flush )
             {
@@ -226,8 +226,8 @@ private void writeMessagesInEventLoop( Message message1, ResponseHandler handler
     {
         channel.eventLoop().execute( () ->
         {
-            messageDispatcher.queue( handler1 );
-            messageDispatcher.queue( handler2 );
+            messageDispatcher.enqueue( handler1 );
+            messageDispatcher.enqueue( handler2 );
 
             channel.write( message1, channel.voidPromise() );
 

driver/src/main/java/org/neo4j/driver/internal/async/inbound/InboundMessageDispatcher.java

@@ -28,6 +28,7 @@
 import org.neo4j.driver.internal.handlers.ResetResponseHandler;
 import org.neo4j.driver.internal.logging.ChannelActivityLogger;
 import org.neo4j.driver.internal.messaging.ResponseMessageHandler;
+import org.neo4j.driver.internal.spi.AutoReadManagingResponseHandler;
 import org.neo4j.driver.internal.spi.ResponseHandler;
 import org.neo4j.driver.internal.util.ErrorUtil;
 import org.neo4j.driver.v1.Logger;
@@ -47,13 +48,15 @@ public class InboundMessageDispatcher implements ResponseMessageHandler
     private Throwable currentError;
     private boolean fatalErrorOccurred;
 
+    private AutoReadManagingResponseHandler autoReadManagingHandler;
+
     public InboundMessageDispatcher( Channel channel, Logging logging )
     {
         this.channel = requireNonNull( channel );
         this.log = new ChannelActivityLogger( channel, logging, getClass() );
     }
 
-    public void queue( ResponseHandler handler )
+    public void enqueue( ResponseHandler handler )
     {
         if ( fatalErrorOccurred )
         {
@@ -62,6 +65,7 @@ public void queue( ResponseHandler handler )
         else
         {
             handlers.add( handler );
+            updateAutoReadManagingHandlerIfNeeded( handler );
         }
     }
 
@@ -74,7 +78,7 @@ public int queuedHandlersCount()
     public void handleSuccessMessage( Map<String,Value> meta )
     {
         log.debug( "S: SUCCESS %s", meta );
-        ResponseHandler handler = handlers.remove();
+        ResponseHandler handler = removeHandler();
         handler.onSuccess( meta );
     }
 
@@ -109,10 +113,10 @@ public void handleFailureMessage( String code, String message )
         }
 
         // write a RESET to "acknowledge" the failure
-        queue( new ResetResponseHandler( this ) );
+        enqueue( new ResetResponseHandler( this ) );
         channel.writeAndFlush( RESET, channel.voidPromise() );
 
-        ResponseHandler handler = handlers.remove();
+        ResponseHandler handler = removeHandler();
         handler.onFailure( currentError );
     }
 
@@ -121,7 +125,7 @@ public void handleIgnoredMessage()
     {
         log.debug( "S: IGNORED" );
 
-        ResponseHandler handler = handlers.remove();
+        ResponseHandler handler = removeHandler();
 
         Throwable error;
         if ( currentError != null )
@@ -145,7 +149,7 @@ public void handleFatalError( Throwable error )
 
         while ( !handlers.isEmpty() )
         {
-            ResponseHandler handler = handlers.remove();
+            ResponseHandler handler = removeHandler();
             handler.onFailure( currentError );
         }
     }
@@ -165,4 +169,44 @@ public boolean fatalErrorOccurred()
         return fatalErrorOccurred;
     }
 
+    /**
+     * <b>Visible for testing</b>
+     */
+    AutoReadManagingResponseHandler autoReadManagingHandler()
+    {
+        return autoReadManagingHandler;
+    }
+
+    private ResponseHandler removeHandler()
+    {
+        ResponseHandler handler = handlers.remove();
+        if ( handler == autoReadManagingHandler )
+        {
+            // the auto-read managing handler is being removed
+            // make sure this dispatcher does not hold on to a removed handler
+            updateAutoReadManagingHandler( null );
+        }
+        return handler;
+    }
+
+    private void updateAutoReadManagingHandlerIfNeeded( ResponseHandler handler )
+    {
+        if ( handler instanceof AutoReadManagingResponseHandler )
+        {
+            updateAutoReadManagingHandler( (AutoReadManagingResponseHandler) handler );
+        }
+    }
+
+    private void updateAutoReadManagingHandler( AutoReadManagingResponseHandler newHandler )
+    {
+        if ( autoReadManagingHandler != null )
+        {
+            // there already exists a handler that manages channel's auto-read
+            // make it stop because new managing handler is being added and there should only be a single such handler
+            autoReadManagingHandler.disableAutoReadManagement();
+            // restore the default value of auto-read
+            channel.config().setAutoRead( true );
+        }
+        autoReadManagingHandler = newHandler;
+    }
 }

driver/src/main/java/org/neo4j/driver/internal/async/pool/NettyChannelHealthChecker.java

@@ -103,7 +103,7 @@ private boolean hasBeenIdleForTooLong( Channel channel )
     private Future<Boolean> ping( Channel channel )
     {
         Promise<Boolean> result = channel.eventLoop().newPromise();
-        messageDispatcher( channel ).queue( new PingResponseHandler( result, channel, log ) );
+        messageDispatcher( channel ).enqueue( new PingResponseHandler( result, channel, log ) );
         channel.writeAndFlush( ResetMessage.RESET, channel.voidPromise() );
         return result;
     }

driver/src/main/java/org/neo4j/driver/internal/handlers/PullAllResponseHandler.java

@@ -27,8 +27,8 @@
 import java.util.concurrent.CompletionStage;
 
 import org.neo4j.driver.internal.InternalRecord;
+import org.neo4j.driver.internal.spi.AutoReadManagingResponseHandler;
 import org.neo4j.driver.internal.spi.Connection;
-import org.neo4j.driver.internal.spi.ResponseHandler;
 import org.neo4j.driver.internal.util.Futures;
 import org.neo4j.driver.internal.util.Iterables;
 import org.neo4j.driver.internal.util.MetadataExtractor;
@@ -44,7 +44,7 @@
 import static org.neo4j.driver.internal.util.Futures.completedWithNull;
 import static org.neo4j.driver.internal.util.Futures.failedFuture;
 
-public abstract class PullAllResponseHandler implements ResponseHandler
+public abstract class PullAllResponseHandler implements AutoReadManagingResponseHandler
 {
     private static final Queue<Record> UNINITIALIZED_RECORDS = Iterables.emptyQueue();
 
@@ -59,6 +59,7 @@ public abstract class PullAllResponseHandler implements ResponseHandler
     // initialized lazily when first record arrives
     private Queue<Record> records = UNINITIALIZED_RECORDS;
 
+    private boolean autoReadManagementEnabled = true;
     private boolean finished;
     private Throwable failure;
     private ResultSummary summary;
@@ -131,6 +132,12 @@ public synchronized void onRecord( Value[] fields )
         }
     }
 
+    @Override
+    public synchronized void disableAutoReadManagement()
+    {
+        autoReadManagementEnabled = false;
+    }
+
     public synchronized CompletionStage<Record> peekAsync()
     {
         Record record = records.peek();
@@ -211,7 +218,7 @@ else if ( finished )
                 // neither SUCCESS nor FAILURE message has arrived, register future to be notified when it arrives
                 // future will be completed with null on SUCCESS and completed with Throwable on FAILURE
                 // enable auto-read, otherwise we might not read SUCCESS/FAILURE if records are not consumed
-                connection.enableAutoRead();
+                enableAutoRead();
                 failureFuture = new CompletableFuture<>();
             }
             return failureFuture;
@@ -236,7 +243,7 @@ private void enqueueRecord( Record record )
             // more than high watermark records are already queued, tell connection to stop auto-reading from network
             // this is needed to deal with slow consumers, we do not want to buffer all records in memory if they are
             // fetched from network faster than consumed
-            connection.disableAutoRead();
+            disableAutoRead();
         }
     }
 
@@ -248,7 +255,7 @@ private Record dequeueRecord()
         {
             // less than low watermark records are now available in the buffer, tell connection to pre-fetch more
             // and populate queue with new records from network
-            connection.enableAutoRead();
+            enableAutoRead();
         }
 
         return record;
@@ -321,4 +328,20 @@ private ResultSummary extractResultSummary( Map<String,Value> metadata )
         long resultAvailableAfter = runResponseHandler.resultAvailableAfter();
         return metadataExtractor.extractSummary( statement, connection, resultAvailableAfter, metadata );
     }
+
+    private void enableAutoRead()
+    {
+        if ( autoReadManagementEnabled )
+        {
+            connection.enableAutoRead();
+        }
+    }
+
+    private void disableAutoRead()
+    {
+        if ( autoReadManagementEnabled )
+        {
+            connection.disableAutoRead();
+        }
+    }
 }

driver/src/main/java/org/neo4j/driver/internal/messaging/v1/BoltProtocolV1.java

@@ -84,7 +84,7 @@ public void initializeChannel( String userAgent, Map<String,Value> authToken, Ch
         InitMessage message = new InitMessage( userAgent, authToken );
         InitResponseHandler handler = new InitResponseHandler( channelInitializedPromise );
 
-        messageDispatcher( channel ).queue( handler );
+        messageDispatcher( channel ).enqueue( handler );
         channel.writeAndFlush( message, channel.voidPromise() );
     }
 

driver/src/main/java/org/neo4j/driver/internal/messaging/v3/BoltProtocolV3.java

@@ -80,7 +80,7 @@ public void initializeChannel( String userAgent, Map<String,Value> authToken, Ch
         HelloMessage message = new HelloMessage( userAgent, authToken );
         HelloResponseHandler handler = new HelloResponseHandler( channelInitializedPromise );
 
-        messageDispatcher( channel ).queue( handler );
+        messageDispatcher( channel ).enqueue( handler );
         channel.writeAndFlush( message, channel.voidPromise() );
     }
 

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

@@ -0,0 +1,38 @@
+/*
+ * Copyright (c) 2002-2018 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.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.spi;
+
+import org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher;
+
+/**
+ * A type of {@link ResponseHandler handler} that manages auto-read of the underlying connection using {@link Connection#enableAutoRead()} and
+ * {@link Connection#disableAutoRead()}.
+ * <p>
+ * Implementations can use auto-read management to apply network-level backpressure when receiving a stream of records.
+ * There should only be a single such handler active for a connection at one point in time. Otherwise, handlers can interfere and turn on/off auto-read
+ * racing with each other. {@link InboundMessageDispatcher} is responsible for tracking these handlers and disabling auto-read management to maintain just
+ * a single auto-read managing handler per connection.
+ */
+public interface AutoReadManagingResponseHandler extends ResponseHandler
+{
+    /**
+     * Tell this handler that it should stop changing auto-read setting for the connection.
+     */
+    void disableAutoReadManagement();
+}

driver/src/test/java/org/neo4j/driver/internal/async/DirectConnectionTest.java

@@ -623,10 +623,10 @@ private static class ThreadTrackingInboundMessageDispatcher extends InboundMessa
         }
 
         @Override
-        public void queue( ResponseHandler handler )
+        public void enqueue( ResponseHandler handler )
         {
             queueThreadNames.add( Thread.currentThread().getName() );
-            super.queue( handler );
+            super.enqueue( handler );
         }
 
     }

driver/src/test/java/org/neo4j/driver/internal/async/HandshakeCompletedListenerTest.java

@@ -113,7 +113,7 @@ private void testWritingOfInitializationMessage( int protocolVersion, Message ex
         listener.operationComplete( handshakeCompletedPromise );
         assertTrue( channel.finish() );
 
-        verify( messageDispatcher ).queue( any( handlerType ) );
+        verify( messageDispatcher ).enqueue( any( handlerType ) );
         Object outboundMessage = channel.readOutbound();
         assertEquals( expectedMessage, outboundMessage );
     }