TransactionContextExecutor extends Observable<T> { public TransactionContextExecutor<T> withSingleTransaction() { return new TransactionContextExecutor<>(SINGLE_TRANSACTION_TRANSACTION_CONTEXT, provider, connectionConsumer); } TransactionContextExecutor(
    TransactionContext transactionContext,
    ConnectionProvider provider,
    ConnectionConsumer<T> connectionConsumer
  ); TransactionContextExecutor<T> withAutoCommit(); TransactionContextExecutor<T> withSingleTransaction(); TransactionContextExecutor<T> withTransactionPerEvent(); static final AutoCommitTransactionContext AUTO_COMMIT_TRANSACTION_CONTEXT; static final SingleTransactionTransactionContext SINGLE_TRANSACTION_TRANSACTION_CONTEXT; static final TransactionPerEventTransactionContext TRANSACTION_PER_EVENT_TRANSACTION_CONTEXT; }
@Test public void singleTransactionShouldCallCommitOnce() throws Exception { Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(null, null)) .withSingleTransaction() .toBlocking().subscribe(o -> {}, throwable -> {}); verify(c, times(1)).commit(); }
@Test public void singleTransactionShouldNotCallRollback() throws Exception { Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(null, null)) .withSingleTransaction() .toBlocking().subscribe(o -> {}, throwable -> {}); verify(c, never()).rollback(); }
@Test public void singleTransactionShouldNotCallCommitOnceOnError() throws Exception { Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(null, null).concatWith(error(new Exception())).concatWith(just(null))) .withSingleTransaction() .toBlocking().subscribe(o -> {}, throwable -> {}); verify(c, never()).commit(); }
@Test public void singleTransactionShouldCallRollbackOnceOnErrorAndNotCallCommit() throws Exception { Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(null, null).concatWith(error(new Exception())).concatWith(just(null))) .withSingleTransaction() .toBlocking().subscribe(o -> {}, throwable -> {}); verify(c, times(1)).rollback(); }
@Test public void singleTransactionShouldCloseConnection() throws Exception { Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(null, null)) .withSingleTransaction() .toBlocking().subscribe(o -> {}, throwable -> {}); verify(c, times(1)).close(); }
@Test public void singleTransactionShouldSetAutoCommitFalse() throws Exception { Connection c = mock(Connection.class); ConnectionPool.of(c).execute(connection -> just(null)).withSingleTransaction().toBlocking().last(); verify(c, times(1)).setAutoCommit(anyBoolean()); verify(c, times(1)).setAutoCommit(eq(false)); }
@Test public void singleTransactionShouldReturnAllValues() throws Exception { TestSubscriber<Object> t = new TestSubscriber<>(); Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(1, 2)) .withSingleTransaction() .subscribe(t); t.assertValues(1, 2); }
@Test public void singleTransactionShouldReturnAllValuesBeforeError() throws Exception { TestSubscriber<Object> t = new TestSubscriber<>(); Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(1, 2).concatWith(error(new Exception())).concatWith(just(3))) .withSingleTransaction() .subscribe(t); t.assertValues(1, 2); }