InternalRxSession extends AbstractRxQueryRunner implements RxSession { @Override public <T> Publisher<T> readTransaction( RxTransactionWork<? extends Publisher<T>> work ) { return readTransaction( work, TransactionConfig.empty() ); } InternalRxSession( NetworkSession session ); @Override Publisher<RxTransaction> beginTransaction(); @Override Publisher<RxTransaction> beginTransaction( TransactionConfig config ); @Override Publisher<T> readTransaction( RxTransactionWork<? extends Publisher<T>> work ); @Override Publisher<T> readTransaction( RxTransactionWork<? extends Publisher<T>> work, TransactionConfig config ); @Override Publisher<T> writeTransaction( RxTransactionWork<? extends Publisher<T>> work ); @Override Publisher<T> writeTransaction( RxTransactionWork<? extends Publisher<T>> work, TransactionConfig config ); @Override RxResult run(String query, TransactionConfig config ); @Override RxResult run(String query, Map<String,Object> parameters, TransactionConfig config ); @Override RxResult run(Query query); @Override RxResult run(Query query, TransactionConfig config ); @Override Bookmark lastBookmark(); Publisher<Void> reset(); @Override Publisher<T> close();  }
@Test void shouldRetryOnError() throws Throwable { 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 ); Publisher<String> strings = rxSession.readTransaction( t -> Flux.just( "a" ).then( Mono.error( new RuntimeException( "Errored" ) ) ) ); StepVerifier.create( Flux.from( strings ) ) .expectError( RuntimeException.class ) .verify(); verify( session, times( retryCount + 1 ) ).beginTransactionAsync( any( AccessMode.class ), any( TransactionConfig.class ) ); verify( tx, times( retryCount + 1 ) ).rollbackAsync(); }
@Test void shouldObtainResultIfRetrySucceed() throws Throwable { 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 ); AtomicInteger count = new AtomicInteger(); Publisher<String> strings = rxSession.readTransaction( t -> { 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(); verify( session, times( retryCount + 1 ) ).beginTransactionAsync( any( AccessMode.class ), any( TransactionConfig.class ) ); verify( tx, times( retryCount ) ).rollbackAsync(); verify( tx ).commitAsync(); }