@Test
void shouldObtainResultIfRetrySucceed() throws Throwable
{

    // Given
    int retryCount = 2;
    NetworkSession session = mock( NetworkSession.class );
    UnmanagedTransaction tx = mock( UnmanagedTransaction.class );
    when( tx.commitAsync() ).thenReturn( completedWithNull() );
    when( tx.rollbackAsync() ).thenReturn( completedWithNull() );
    when( session.beginTransactionAsync( any( AccessMode.class ), any( TransactionConfig.class ) ) ).thenReturn( completedFuture( tx ) );
    when( session.retryLogic() ).thenReturn( new FixedRetryLogic( retryCount ) );
    InternalRxSession rxSession = new InternalRxSession( session );
    // When
    AtomicInteger count = new AtomicInteger();
    Publisher<String> strings = rxSession.readTransaction( t -> {
        // we fail for the first few retries, and then success on the last run.
        if ( count.getAndIncrement() == retryCount )
        {
            return Flux.just( "a" );
        }
        else
        {
            return Flux.just( "a" ).then( Mono.error( new RuntimeException( "Errored" ) ) );
        }
    } );
    StepVerifier.create( Flux.from( strings ) ).expectNext( "a" ).verifyComplete();
    // Then
    verify( session, times( retryCount + 1 ) ).beginTransactionAsync( any( AccessMode.class ), any( TransactionConfig.class ) );
    verify( tx, times( retryCount ) ).rollbackAsync();
    verify( tx ).commitAsync();
}