Merge pull request vert-x3#259 from EmadAlblueshi/rx-tx-example

Adds transaction management examples for RxJava and RxJava 2

Author: cescoffier

rxjava-1-examples/README.adoc

@@ -141,6 +141,13 @@ operations via the `flatMap` operation and then subscribes to the result.
 
 link:src/main/java/io/vertx/example/rxjava/database/jdbc/Client.java[RxJava jdbc client]
 
+=== Jdbc Manual Transaction Handling
+
+An example showing an Rxified Jdbc api to handle transactions manually then commit if all succeeded or rollback with
+exception propagation to the caller in case of anyone failed.
+
+link:src/main/java/io/vertx/example/rxjava/database/jdbc/Transaction.java[RxJava JDBC Transaction]
+
 === Mongo example
 
 An example showing the Mongo Service Rxified api, after the client connected to Mongo, it chains

rxjava-1-examples/src/main/java/io/vertx/example/rxjava/database/jdbc/Transaction.java

@@ -0,0 +1,63 @@
+package io.vertx.example.rxjava.database.jdbc;
+
+import io.vertx.core.json.JsonArray;
+import io.vertx.core.json.JsonObject;
+import io.vertx.example.util.Runner;
+import io.vertx.rxjava.core.AbstractVerticle;
+import io.vertx.rxjava.ext.jdbc.JDBCClient;
+import rx.Single;
+import rx.exceptions.CompositeException;
+
+/*
+ * @author <a href="mailto:[email protected]">Emad Alblueshi</a>
+ */
+public class Transaction extends AbstractVerticle {
+
+  // Convenience method so you can run it in your IDE
+  public static void main(String[] args) {
+    Runner.runExample(Transaction.class);
+  }
+
+  @Override
+  public void start() throws Exception {
+
+    JsonObject config = new JsonObject().put("url", "jdbc:hsqldb:mem:test?shutdown=true")
+      .put("driver_class", "org.hsqldb.jdbcDriver");
+
+    String sql = "CREATE TABLE colors (" +
+      "id INTEGER GENERATED BY DEFAULT AS IDENTITY(START WITH 1, INCREMENT BY 1) PRIMARY KEY, " +
+      "name VARCHAR(255), " +
+      "datetime TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL)";
+
+    JDBCClient client = JDBCClient.createShared(vertx, config);
+
+    // Connect to the database
+    client
+      .rxGetConnection()
+      .flatMap(conn ->
+        conn
+          // Disable auto commit to handle transaction manually
+          .rxSetAutoCommit(false)
+          // Create table
+          .flatMap(autoCommit -> conn.rxExecute(sql))
+          // Insert colors
+          .flatMap(executed -> conn.rxUpdateWithParams("INSERT INTO colors (name) VALUES (?)", new JsonArray().add("BLACK")))
+          .flatMap(updateResult -> conn.rxUpdateWithParams("INSERT INTO colors (name) VALUES (?)", new JsonArray().add("WHITE")))
+          .flatMap(updateResult -> conn.rxUpdateWithParams("INSERT INTO colors (name) VALUES (?)", new JsonArray().add("PURPLE")))
+          // Commit if all succeeded
+          .flatMap(updateResult -> conn.rxCommit().map(commit -> updateResult))
+          // Rollback if any failed with exception propagation
+          .onErrorResumeNext(ex -> conn.rxRollback()
+            .onErrorResumeNext(ex2 -> Single.error(new CompositeException(ex, ex2)))
+            .flatMap(ignore -> Single.error(ex))
+          )
+          // Get colors if all succeeded
+          .flatMap(updateResult -> conn.rxQuery("SELECT * FROM colors"))
+          // Close the connection regardless succeeded or failed
+          .doAfterTerminate(conn::close)
+      ).subscribe(resultSet -> {
+      // Subscribe to get the final result
+      System.out.println("Results : " + resultSet.getRows());
+    }, Throwable::printStackTrace);
+  }
+}

rxjava-2-examples/README.adoc

@@ -142,6 +142,13 @@ operations via the `flatMap` operation and then subscribes to the result.
 
 link:src/main/java/io/vertx/example/reactivex/database/jdbc/Client.java[RxJava 2 JDBC client]
 
+=== Jdbc Manual Transaction Handling
+
+An example showing an Rxified Jdbc api to handle transactions manually then commit if all succeeded or rollback with
+exception propagation to the caller in case of anyone failed.
+
+link:src/main/java/io/vertx/example/reactivex/database/jdbc/Transaction.java[RxJava Jdbc Transaction]
+
 === Mongo example
 
 An example showing the Mongo Service Rxified api, after the client connected to Mongo, it chains

rxjava-2-examples/src/main/java/io/vertx/example/reactivex/database/jdbc/Transaction.java

@@ -0,0 +1,66 @@
+package io.vertx.example.reactivex.database.jdbc;
+
+import io.reactivex.Single;
+import io.reactivex.exceptions.CompositeException;
+import io.vertx.core.json.JsonArray;
+import io.vertx.core.json.JsonObject;
+import io.vertx.example.util.Runner;
+import io.vertx.reactivex.core.AbstractVerticle;
+import io.vertx.reactivex.ext.jdbc.JDBCClient;
+
+/*
+ * @author <a href="mailto:[email protected]">Emad Alblueshi</a>
+ */
+public class Transaction extends AbstractVerticle {
+
+  // Convenience method so you can run it in your IDE
+  public static void main(String[] args) {
+    Runner.runExample(Transaction.class);
+  }
+
+  @Override
+  public void start() throws Exception {
+
+    JsonObject config = new JsonObject().put("url", "jdbc:hsqldb:mem:test?shutdown=true")
+      .put("driver_class", "org.hsqldb.jdbcDriver");
+
+    String sql = "CREATE TABLE colors (" +
+      "id INTEGER GENERATED BY DEFAULT AS IDENTITY(START WITH 1, INCREMENT BY 1) PRIMARY KEY, " +
+      "name VARCHAR(255), " +
+      "datetime TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL)";
+
+    JDBCClient client = JDBCClient.createShared(vertx, config);
+
+    // Connect to the database
+    client
+      .rxGetConnection()
+      .flatMap(conn ->
+        conn
+          // Disable auto commit to handle transaction manually
+          .rxSetAutoCommit(false)
+          // Switch from Completable to default Single value
+          .toSingleDefault(false)
+          // Create table
+          .flatMap(autoCommit -> conn.rxExecute(sql).toSingleDefault(true))
+          // Insert colors
+          .flatMap(executed -> conn.rxUpdateWithParams("INSERT INTO colors (name) VALUES (?)", new JsonArray().add("BLACK")))
+          .flatMap(updateResult -> conn.rxUpdateWithParams("INSERT INTO colors (name) VALUES (?)", new JsonArray().add("WHITE")))
+          .flatMap(updateResult -> conn.rxUpdateWithParams("INSERT INTO colors (name) VALUES (?)", new JsonArray().add("PURPLE")))
+          // commit if all succeeded
+          .flatMap(updateResult -> conn.rxCommit().toSingleDefault(true).map(commit -> updateResult))
+          // Rollback if any failed with exception propagation
+          .onErrorResumeNext(ex -> conn.rxRollback()
+            .toSingleDefault(true)
+            .onErrorResumeNext(ex2 -> Single.error(new CompositeException(ex, ex2)))
+            .flatMap(ignore -> Single.error(ex))
+          )
+          // Get colors if all succeeded
+          .flatMap(updateResult -> conn.rxQuery("SELECT * FROM colors"))
+          // close the connection regardless succeeded or failed
+          .doAfterTerminate(conn::close)
+      ).subscribe(resultSet -> {
+      // Subscribe to get the final result
+      System.out.println("Results : " + resultSet.getRows());
+    }, Throwable::printStackTrace);
+  }
+}