AutoPullResponseHandler extends BasicPullResponseHandler implements PullAllResponseHandler { public synchronized CompletionStage<Record> nextAsync() { return peekAsync().thenApply( ignore -> dequeueRecord() ); } AutoPullResponseHandler(Query query, RunResponseHandler runResponseHandler, Connection connection, MetadataExtractor metadataExtractor,
                                   PullResponseCompletionListener completionListener, long fetchSize ); synchronized CompletionStage<Record> peekAsync(); synchronized CompletionStage<Record> nextAsync(); synchronized CompletionStage<ResultSummary> consumeAsync(); synchronized CompletionStage<List<T>> listAsync( Function<Record,T> mapFunction ); @Override synchronized CompletionStage<Throwable> pullAllFailureAsync(); @Override void prePopulateRecords();  }
@Test void shouldRestartRequestingWhenMinimumWatermarkMet() { Connection connection = connectionMock(); InOrder inOrder = Mockito.inOrder( connection ); AutoPullResponseHandler handler = newHandler( new Query( "RETURN 1" ), connection, 7 ); Map<String,Value> metaData = new HashMap<>( 1 ); metaData.put( "has_more", BooleanValue.TRUE ); inOrder.verify( connection ).writeAndFlush( any( PullMessage.class ), any() ); handler.onRecord( values( 1 ) ); handler.onRecord( values( 2 ) ); handler.onRecord( values( 3 ) ); handler.onRecord( values( 4 ) ); handler.onRecord( values( 5 ) ); handler.onSuccess( metaData ); verify( connection, times( 1 ) ).writeAndFlush( any( PullMessage.class ), any() ); handler.nextAsync(); handler.nextAsync(); handler.nextAsync(); inOrder.verify( connection ).writeAndFlush( any( PullMessage.class ), any() ); }
@Test void shouldFunctionWhenHighAndLowWatermarksAreEqual() { Connection connection = connectionMock(); InOrder inOrder = Mockito.inOrder( connection ); AutoPullResponseHandler handler = newHandler( new Query( "RETURN 1" ), connection, 1 ); Map<String,Value> metaData = new HashMap<>( 1 ); metaData.put( "has_more", BooleanValue.TRUE ); inOrder.verify( connection ).writeAndFlush( any( PullMessage.class ), any() ); handler.onRecord( values( 1 ) ); handler.onSuccess( metaData ); inOrder.verify( connection, never() ).writeAndFlush( any(), any() ); handler.nextAsync(); inOrder.verify( connection ).writeAndFlush( any( PullMessage.class ), any() ); }