Transaction management for SQLClient

[tsegismont]

When using an Rxified SQLClient, it's not obvious for everyone how to properly manage a connection. Proper transaction management is even worse.

It would help to have utility methods to manage a connection, a transaction, or both.

Let's take an example:

public class JDBCHelper {

  static <T> Flowable<T> withConnectionFlowable(SQLClient sqlClient, Function<SQLConnection, Flowable<T>> flowable) {
    return sqlClient.rxGetConnection().flatMapPublisher(sqlConnection -> {
      return flowable.apply(sqlConnection).doFinally(sqlConnection::close);
    });
  }

  static <T> Flowable<T> inTransactionFlowable(SQLClient sqlClient, Function<SQLConnection, Flowable<T>> flowable) {
    return withConnectionFlowable(sqlClient, flowable);
  }

  static <T> Flowable<T> inTransactionFlowable(SQLConnection sqlConnection, Function<SQLConnection, Flowable<T>> flowable) {
    return sqlConnection.rxSetAutoCommit(false).<T>toFlowable()
      .concatWith(flowable.apply(sqlConnection))
      .concatWith(sqlConnection.rxCommit().toFlowable())
      .onErrorResumeNext(throwable -> {
        return sqlConnection.rxRollback().onErrorComplete().andThen(Flowable.error(throwable));
      }).concatWith(sqlConnection.rxSetAutoCommit(true).toFlowable());
  }
}

Now it's easy to write such kind of flow:

    JDBCHelper.withConnectionFlowable(client, conn -> {
      Flowable<JsonArray> setupDb = JDBCHelper.inTransactionFlowable(conn, sqlConnection -> {
        return conn.rxUpdateWithParams("INSERT INTO colors (name) VALUES (?)", new JsonArray().add("BLACK")).toCompletable().toFlowable()
          .concatWith(conn.rxUpdateWithParams("INSERT INTO colors (name) VALUES (?)", new JsonArray().add("WHITE")).toCompletable().toFlowable())
          .concatWith(conn.rxUpdateWithParams("INSERT INTO colors (name) VALUES (?)", new JsonArray().add("PURPLE")).toCompletable().toFlowable());
      }).ignoreElements().toFlowable();
      return setupDb
        .concatWith(conn.rxQuery("SELECT * FROM colors").flatMapPublisher(resultSet -> Flowable.fromIterable(resultSet.getResults())));
    }).subscribe(System.out::println, Throwable::printStackTrace);

Of course, we'd need other helper methods for Single, Completable and Maybe.

Also, it's not clear to me where we'd add this code. In this repo? In vertx-sql-common? Should we use utility classes as in the example above? Or should we add these methods to Rxified SQLClient classes? In this case, how?

[vietj]

the appropriate location in an helper class in vertx-rx-java2/src/main/java/io/vertx/reactivex/ext/sql

[tsegismont]

Thanks @vietj ! Any comment on the proposal itself?

[vietj]

it sounds a better alternative to my initial proposal I did for 3.6 to ease finally like semantics on Flowable. So I'm happy you made it.

I think that withConnectionFlowable should rather be in the normal RxHelper using instead the Closeable interface since SqlConnection implements this interface (with a better name).

[tsegismont]

So I'm happy you made it.

Well, just threw some code here so far, not tested yet :)

I think that withConnectionFlowable should rather be in the normal RxHelper using instead the Closeable interface since SqlConnection implements this interface (with a better name).

👍

[tsegismont]

@vietj as discussed off-list, narrowing down the scope of this issue to transaction management only. We will address Closeable / AutoCloseable separately. The PR will have quite a lot of changes anyway :)

[vietj]

I guess we can close this PR ?

[tsegismont]

👍