InternalRxResult implements RxResult { @Override public Publisher<List<String>> keys() { return Mono.defer( () -> Mono.fromCompletionStage( getCursorFuture() ).map( RxResultCursor::keys ) .onErrorMap( Futures::completionExceptionCause ) ); } InternalRxResult(Supplier<CompletionStage<RxResultCursor>> cursorFuture ); @Override Publisher<List<String>> keys(); @Override Publisher<Record> records(); @Override Publisher<ResultSummary> consume();  }
@Test void shouldObtainKeys() { RxResultCursor cursor = mock( RxResultCursorImpl.class ); RxResult rxResult = newRxResult( cursor ); List<String> keys = Arrays.asList( "one", "two", "three" ); when( cursor.keys() ).thenReturn( keys ); StepVerifier.create( Flux.from( rxResult.keys() ) ) .expectNext( Arrays.asList( "one", "two", "three" ) ) .verifyComplete(); }
@Test void shouldErrorWhenFailedObtainKeys() { RuntimeException error = new RuntimeException( "Failed to obtain cursor" ); InternalRxResult rxResult = newRxResult( error ); StepVerifier.create( Flux.from( rxResult.keys() ) ) .expectErrorMatches( isEqual( error ) ) .verify(); }
@Test void shouldCancelKeys() { RxResultCursor cursor = mock( RxResultCursorImpl.class ); RxResult rxResult = newRxResult( cursor ); List<String> keys = Arrays.asList( "one", "two", "three" ); when( cursor.keys() ).thenReturn( keys ); StepVerifier.create( Flux.from( rxResult.keys() ).limitRate( 1 ).take( 1 ) ) .expectNext( Arrays.asList( "one", "two", "three" ) ) .verifyComplete(); }