RxResultCursorImpl implements RxResultCursor { @Override public void installRecordConsumer( BiConsumer<Record,Throwable> recordConsumer ) { if ( resultConsumed ) { throw newResultConsumedError(); } if ( consumerStatus.isInstalled() ) { return; } consumerStatus = recordConsumer == DISCARD_RECORD_CONSUMER ? DISCARD_INSTALLED : INSTALLED; pullHandler.installRecordConsumer( recordConsumer ); assertRunCompletedSuccessfully(); } RxResultCursorImpl(RunResponseHandler runHandler, PullResponseHandler pullHandler );  RxResultCursorImpl(Throwable runError, RunResponseHandler runHandler, PullResponseHandler pullHandler ); @Override List<String> keys(); @Override void installRecordConsumer( BiConsumer<Record,Throwable> recordConsumer ); @Override void request( long n ); @Override void cancel(); @Override CompletionStage<Throwable> discardAllFailureAsync(); @Override CompletionStage<Throwable> pullAllFailureAsync(); @Override CompletionStage<ResultSummary> summaryAsync(); @Override boolean isDone();  }
@Test void shouldInstallRecordConsumerAndReportError() throws Throwable { RuntimeException error = new RuntimeException( "Hi" ); BiConsumer recordConsumer = mock( BiConsumer.class ); RunResponseHandler runHandler = newRunResponseHandler( error ); PullResponseHandler pullHandler = new ListBasedPullHandler(); RxResultCursor cursor = new RxResultCursorImpl( error, runHandler, pullHandler ); cursor.installRecordConsumer( recordConsumer ); verify( recordConsumer ).accept( null, error ); verifyNoMoreInteractions( recordConsumer ); }
@Test void shouldOnlyInstallRecordConsumerOnce() throws Throwable { RunResponseHandler runHandler = newRunResponseHandler(); PullResponseHandler pullHandler = mock( PullResponseHandler.class ); RxResultCursor cursor = new RxResultCursorImpl( runHandler, pullHandler ); cursor.installRecordConsumer( DISCARD_RECORD_CONSUMER ); cursor.installRecordConsumer( DISCARD_RECORD_CONSUMER ); verify( pullHandler ).installRecordConsumer( any() ); }