@Test public void testTx() throws InterruptedException { Database db = db(3); Flowable<Tx<?>> transaction = db .update("update person set score=-3 where name='FRED'") .transaction(); transaction .doOnCancel(() -> log.debug("disposing")) .doOnNext(DatabaseTest::println) .flatMap(tx -> { log.debug("flatmapping"); return tx .update("update person set score = -4 where score = -3") .countsOnly() .doOnSubscribe(s -> log.debug("subscribed")) .doOnNext(num -> log.debug("num=" + num)); }) .test() .awaitDone(TIMEOUT_SECONDS, TimeUnit.SECONDS) .assertNoErrors() .assertValue(1) .assertComplete(); }
public TransactedBuilder tx(@Nonnull Tx<?> tx) { Preconditions.checkNotNull(tx, "tx cannot be null"); TxImpl<?> t = (TxImpl<?>) tx; TransactedConnection c = t.connection().fork(); return new TransactedBuilder(c, this); }
Database implements AutoCloseable { public TransactedBuilder tx(@Nonnull Tx<?> tx) { Preconditions.checkNotNull(tx, "tx cannot be null"); TxImpl<?> t = (TxImpl<?>) tx; TransactedConnection c = t.connection().fork(); return new TransactedBuilder(c, this); } }
Database implements AutoCloseable { public TransactedBuilder tx(@Nonnull Tx<?> tx) { Preconditions.checkNotNull(tx, "tx cannot be null"); TxImpl<?> t = (TxImpl<?>) tx; TransactedConnection c = t.connection().fork(); return new TransactedBuilder(c, this); } private  Database(@Nonnull Pool<Connection> pool, @Nonnull Action onClose); }
Database implements AutoCloseable { public TransactedBuilder tx(@Nonnull Tx<?> tx) { Preconditions.checkNotNull(tx, "tx cannot be null"); TxImpl<?> t = (TxImpl<?>) tx; TransactedConnection c = t.connection().fork(); return new TransactedBuilder(c, this); } private  Database(@Nonnull Pool<Connection> pool, @Nonnull Action onClose); static NonBlockingConnectionPool.Builder<Database> nonBlocking(); static Database from(@Nonnull String url, int maxPoolSize); static Database from(@Nonnull Pool<Connection> pool); static Database from(@Nonnull Pool<Connection> pool, Action closeAction); static Database fromBlocking(@Nonnull ConnectionProvider cp); static Database fromBlocking(@Nonnull DataSource dataSource); static Database test(int maxPoolSize); static Database test(); Single<Connection> connection(); Flowable<Connection> connections(); @Override void close(); CallableBuilder call(@Nonnull String sql); SelectAutomappedBuilder<T> select(@Nonnull Class<T> cls); SelectBuilder select(@Nonnull String sql); UpdateBuilder update(@Nonnull String sql); TransactedBuilder tx(@Nonnull Tx<?> tx); static Object toSentinelIfNull(@Nullable String s); static Object toSentinelIfNull(@Nullable byte[] bytes); static Object clob(@Nullable String s); static Object blob(@Nullable byte[] bytes); Single<Member<Connection>> member(); Single<T> apply(Function<? super Connection, ? extends T> function); Completable apply(Consumer<? super Connection> consumer); }
Database implements AutoCloseable { public TransactedBuilder tx(@Nonnull Tx<?> tx) { Preconditions.checkNotNull(tx, "tx cannot be null"); TxImpl<?> t = (TxImpl<?>) tx; TransactedConnection c = t.connection().fork(); return new TransactedBuilder(c, this); } private  Database(@Nonnull Pool<Connection> pool, @Nonnull Action onClose); static NonBlockingConnectionPool.Builder<Database> nonBlocking(); static Database from(@Nonnull String url, int maxPoolSize); static Database from(@Nonnull Pool<Connection> pool); static Database from(@Nonnull Pool<Connection> pool, Action closeAction); static Database fromBlocking(@Nonnull ConnectionProvider cp); static Database fromBlocking(@Nonnull DataSource dataSource); static Database test(int maxPoolSize); static Database test(); Single<Connection> connection(); Flowable<Connection> connections(); @Override void close(); CallableBuilder call(@Nonnull String sql); SelectAutomappedBuilder<T> select(@Nonnull Class<T> cls); SelectBuilder select(@Nonnull String sql); UpdateBuilder update(@Nonnull String sql); TransactedBuilder tx(@Nonnull Tx<?> tx); static Object toSentinelIfNull(@Nullable String s); static Object toSentinelIfNull(@Nullable byte[] bytes); static Object clob(@Nullable String s); static Object blob(@Nullable byte[] bytes); Single<Member<Connection>> member(); Single<T> apply(Function<? super Connection, ? extends T> function); Completable apply(Consumer<? super Connection> consumer); static final Object NULL_CLOB; static final Object NULL_NUMBER; static final Object NULL_BLOB; }