Make reactive subscription request with Long.MAX_VALUE unbounded (#1003)

* Make reactive subscription request with Long.MAX_VALUE unbounded

This update ensures that unbounded reactive request is translated to unbounded Bolt request.

* Update ListBasedPullHandler test class to support unbounded request

Author: injectives

driver/src/main/java/org/neo4j/driver/internal/cursor/RxResultCursorImpl.java

@@ -88,6 +88,10 @@ public void installRecordConsumer( BiConsumer<Record,Throwable> recordConsumer )
     @Override
     public void request( long n )
     {
+        if ( n == Long.MAX_VALUE )
+        {
+            n = -1;
+        }
         pullHandler.request( n );
     }
 

driver/src/test/java/org/neo4j/driver/internal/cursor/RxResultCursorImplTest.java

@@ -122,6 +122,21 @@ void shouldPull()
         verify( pullHandler ).request( 100 );
     }
 
+    @Test
+    void shouldPullUnboundedOnLongMax()
+    {
+        // Given
+        RunResponseHandler runHandler = newRunResponseHandler();
+        PullResponseHandler pullHandler = mock( PullResponseHandler.class );
+        RxResultCursor cursor = new RxResultCursorImpl( runHandler, pullHandler );
+
+        // When
+        cursor.request( Long.MAX_VALUE );
+
+        // Then
+        verify( pullHandler ).request( -1 );
+    }
+
     @Test
     void shouldCancel()
     {

driver/src/test/java/org/neo4j/driver/internal/reactive/util/ListBasedPullHandler.java

@@ -22,15 +22,15 @@
 import java.util.Map;
 
 import org.neo4j.driver.Query;
+import org.neo4j.driver.Record;
+import org.neo4j.driver.Value;
 import org.neo4j.driver.internal.handlers.PullResponseCompletionListener;
 import org.neo4j.driver.internal.handlers.RunResponseHandler;
 import org.neo4j.driver.internal.handlers.pulln.BasicPullResponseHandler;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.util.MetadataExtractor;
 import org.neo4j.driver.internal.util.QueryKeys;
 import org.neo4j.driver.internal.value.BooleanValue;
-import org.neo4j.driver.Record;
-import org.neo4j.driver.Value;
 import org.neo4j.driver.summary.ResultSummary;
 
 import static java.util.Collections.emptyList;
@@ -81,7 +81,7 @@ private ListBasedPullHandler( List<Record> list, Throwable error )
     public void request( long n )
     {
         super.request( n );
-        while ( index < list.size() && n-- > 0 )
+        while ( index < list.size() && (n == -1 || n-- > 0) )
         {
             onRecord( list.get( index++ ).values().toArray( new Value[0] ) );
         }

testkit-backend/src/main/java/neo4j/org/testkit/backend/messages/requests/ResultNext.java

@@ -132,9 +132,10 @@ private CompletionStage<Record> consumeNextRecordOrCompletionSignal( RxBlockingS
 
     private long getFetchSize( RxResultHolder resultHolder )
     {
-        return resultHolder.getSessionHolder().getConfig()
-                           .fetchSize()
-                           .orElse( resultHolder.getSessionHolder().getDriverHolder().getConfig().fetchSize() );
+        long fetchSize = resultHolder.getSessionHolder().getConfig()
+                                     .fetchSize()
+                                     .orElse( resultHolder.getSessionHolder().getDriverHolder().getConfig().fetchSize() );
+        return fetchSize == -1 ? Long.MAX_VALUE : fetchSize;
     }
 
     private neo4j.org.testkit.backend.messages.responses.Record createResponse( Record record )

testkit-backend/src/main/java/neo4j/org/testkit/backend/messages/requests/StartTest.java

@@ -80,12 +80,6 @@ public class StartTest implements TestkitRequest
                                              skipMessage );
         REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.Routing[^.]+\\.test_should_retry_write_until_success_with_leader_change_using_tx_function$",
                                              skipMessage );
-        skipMessage = "Fetch size -1 not supported";
-        REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.NoRouting[^.]+\\.test_should_pull_all_when_fetch_is_minus_one_using_driver_configuration$", skipMessage );
-        REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestIterationSessionRun\\.test_all$", skipMessage );
-        REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestIterationSessionRun\\.test_all_slow_connection$", skipMessage );
-        REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestIterationTxRun\\.test_all$", skipMessage );
-        REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestIterationTxRun\\.test_all_slow_connection$", skipMessage );
         REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.TestSessionRun\\.test_discard_on_session_close_unfinished_result$",
                                              "Does not support partially consumed state" );
         REACTIVE_SKIP_PATTERN_TO_REASON.put( "^.*\\.NoRouting[^.]+\\.test_should_error_on_database_shutdown_using_tx_run$", "Session close throws error" );