TransactionContextExecutor extends Observable<T> { public TransactionContextExecutor<T> withAutoCommit() { return new TransactionContextExecutor<>(AUTO_COMMIT_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 autoCommitShouldSetAutoCommitTrue() throws Exception { Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(null, null)) .withAutoCommit() .toBlocking().subscribe(o -> {}, throwable -> {}); verify(c, times(1)).setAutoCommit(anyBoolean()); verify(c, times(1)).setAutoCommit(eq(true)); }
@Test public void autoCommitShouldReturnAllValues() throws Exception { TestSubscriber<Object> t = new TestSubscriber<>(); Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(1, 2)) .withAutoCommit() .subscribe(t); t.assertValues(1, 2); }
@Test public void autoCommitShouldReturnAllValuesBeforeError() 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))) .withAutoCommit() .subscribe(t); t.assertValues(1, 2); }
@Test public void autoCommitShouldNotCallCommit() throws Exception { TestSubscriber<Object> t = new TestSubscriber<>(); Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(null, null)) .withAutoCommit() .toBlocking().subscribe(o -> {}, throwable -> {}); verify(c, never()).commit(); }
@Test public void autoCommitShouldNotCallRollbackOnError() throws Exception { TestSubscriber<Object> t = new TestSubscriber<>(); Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(null, null).concatWith(error(new Exception()))) .withAutoCommit() .toBlocking().subscribe(o -> {}, throwable -> {}); verify(c, never()).rollback(); }
@Test public void autoCommitShouldNotCallRollback() throws Exception { Connection c = mock(Connection.class); ConnectionPool.of(c) .execute(connection -> just(null, null)) .withAutoCommit() .toBlocking().subscribe(o -> {}, throwable -> {}); verify(c, never()).rollback(); }
@Test public void autoCommitShouldCloseConnection() throws Exception { Connection c = mock(Connection.class); ConnectionPool.of(c).execute(connection -> just(null, null)) .withAutoCommit().toBlocking() .subscribe(o -> {}, throwable -> {}); verify(c, times(1)).close(); }