Disable the removed auto-read managing handler

To make sure it does not influence the auto-read status of the channel.
Also restore default value of auto-read for channel when such handler
is removed.

This change is needed to avoid removing a handler that has just
disabled auto-read. This is possible when handler receives a RECORD
that causes it to disable auto-read and then receives a SUCCESS
message, which causes it to be dequeued with auto-read being disabled.

Author: lutovich

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

@@ -265,11 +265,11 @@ private void ackFailureIfNeeded()
     private ResponseHandler removeHandler()
     {
         ResponseHandler handler = handlers.remove();
-        if ( autoReadManagingHandler == handler )
+        if ( handler == autoReadManagingHandler )
         {
-            // handler that is being removed is the auto-read managing handler
+            // the auto-read managing handler is being removed
             // make sure this dispatcher does not hold on to a removed handler
-            autoReadManagingHandler = null;
+            updateAutoReadManagingHandler( null );
         }
         return handler;
     }
@@ -278,15 +278,20 @@ private void updateAutoReadManagingHandlerIfNeeded( ResponseHandler handler )
     {
         if ( handler instanceof AutoReadManagingResponseHandler )
         {
-            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 = (AutoReadManagingResponseHandler) handler;
+            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/test/java/org/neo4j/driver/internal/async/inbound/InboundMessageDispatcherTest.java

@@ -44,6 +44,7 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyBoolean;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
@@ -503,6 +504,25 @@ public void shouldForgetAutoReadManagingHandlerWhenItIsRemoved()
         assertNull( dispatcher.autoReadManagingHandler() );
     }
 
+    @Test
+    public void shouldReEnableAutoReadWhenAutoReadManagingHandlerIsRemoved()
+    {
+        Channel channel = newChannelMock();
+        InboundMessageDispatcher dispatcher = newDispatcher( channel );
+
+        AutoReadManagingResponseHandler handler = mock( AutoReadManagingResponseHandler.class );
+        dispatcher.enqueue( handler );
+        assertEquals( handler, dispatcher.autoReadManagingHandler() );
+        verify( handler, never() ).disableAutoReadManagement();
+        verify( channel.config(), never() ).setAutoRead( anyBoolean() );
+
+        dispatcher.handleSuccessMessage( emptyMap() );
+
+        assertNull( dispatcher.autoReadManagingHandler() );
+        verify( handler ).disableAutoReadManagement();
+        verify( channel.config() ).setAutoRead( anyBoolean() );
+    }
+
     private static void verifyFailure( ResponseHandler handler )
     {
         ArgumentCaptor<Neo4jException> captor = ArgumentCaptor.forClass( Neo4jException.class );