ResultCursorFactoryImpl implements ResultCursorFactory { @Override public CompletionStage<RxResultCursor> rxResult() { connection.writeAndFlush( runMessage, runHandler ); return runHandler.runFuture().thenApply( this::composeRxCursor ); } ResultCursorFactoryImpl(Connection connection, Message runMessage, RunResponseHandler runHandler, PullResponseHandler pullHandler,
                                   PullAllResponseHandler pullAllHandler, boolean waitForRunResponse ); @Override CompletionStage<AsyncResultCursor> asyncResult(); @Override CompletionStage<RxResultCursor> rxResult();  }
@Test void shouldFailRxResultWhenRunFailed() throws Throwable { Throwable error = new RuntimeException( "Hi there" ); ResultCursorFactory cursorFactory = newResultCursorFactory( failedFuture( error ), true ); CompletionStage<RxResultCursor> rxCursorFuture = cursorFactory.rxResult(); CompletionException actual = assertThrows( CompletionException.class, () -> getNow( rxCursorFuture ) ); assertThat( actual.getCause(), equalTo( error ) ); }
@Test void shouldNotFailRxResultEvenWhenRunFailed() throws Throwable { Connection connection = mock( Connection.class ); Throwable error = new RuntimeException( "Hi there" ); ResultCursorFactory cursorFactory = newResultCursorFactory( connection, completedFuture( error ), false ); CompletionStage<RxResultCursor> cursorFuture = cursorFactory.rxResult(); verifyRxRunCompleted( connection, cursorFuture ); }