Simplified transaction API improvements (#574)

Fixes

- #421
- #583
- #584
- #585
- #586

Author: vietj

vertx-db2-client/src/main/java/examples/SqlClientExamples.java

@@ -210,32 +210,38 @@ public void transaction01(Pool pool) {
         SqlConnection conn = res.result();
 
         // Begin the transaction
-        Transaction tx = conn.begin();
-
-        // Various statements
-        conn
-          .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
-          .execute(ar1 -> {
-          if (ar1.succeeded()) {
+        conn.begin(ar0 -> {
+          if (ar0.succeeded()) {
+            Transaction tx = ar0.result();
+            // Various statements
             conn
-              .query("INSERT INTO Users (first_name,last_name) VALUES ('Emad','Alblueshi')")
-              .execute(ar2 -> {
-              if (ar2.succeeded()) {
-                // Commit the transaction
-                tx.commit(ar3 -> {
-                  if (ar3.succeeded()) {
-                    System.out.println("Transaction succeeded");
-                  } else {
-                    System.out.println("Transaction failed " + ar3.cause().getMessage());
-                  }
+              .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
+              .execute(ar1 -> {
+                if (ar1.succeeded()) {
+                  conn
+                    .query("INSERT INTO Users (first_name,last_name) VALUES ('Emad','Alblueshi')")
+                    .execute(ar2 -> {
+                      if (ar2.succeeded()) {
+                        // Commit the transaction
+                        tx.commit(ar3 -> {
+                          if (ar3.succeeded()) {
+                            System.out.println("Transaction succeeded");
+                          } else {
+                            System.out.println("Transaction failed " + ar3.cause().getMessage());
+                          }
+                          // Return the connection to the pool
+                          conn.close();
+                        });
+                      } else {
+                        // Return the connection to the pool
+                        conn.close();
+                      }
+                    });
+                } else {
                   // Return the connection to the pool
                   conn.close();
-                });
-              } else {
-                // Return the connection to the pool
-                conn.close();
-              }
-            });
+                }
+              });
           } else {
             // Return the connection to the pool
             conn.close();
@@ -246,69 +252,62 @@ public void transaction01(Pool pool) {
   }
 
   public void transaction02(Transaction tx) {
-    tx.abortHandler(v -> {
+    tx.completion().onFailure(err -> {
       System.out.println("Transaction failed => rollbacked");
     });
   }
 
   public void transaction03(Pool pool) {
 
     // Acquire a transaction and begin the transaction
-    pool.begin(res -> {
-      if (res.succeeded()) {
-
-        // Get the transaction
-        Transaction tx = res.result();
-
-        // Various statements
-        tx.query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
-          .execute(ar1 -> {
-          if (ar1.succeeded()) {
-            tx.query("INSERT INTO Users (first_name,last_name) VALUES ('Emad','Alblueshi')")
-              .execute(ar2 -> {
-              if (ar2.succeeded()) {
-                // Commit the transaction
-                // the connection will automatically return to the pool
-                tx.commit(ar3 -> {
-                  if (ar3.succeeded()) {
-                    System.out.println("Transaction succeeded");
-                  } else {
-                    System.out.println("Transaction failed " + ar3.cause().getMessage());
-                  }
-                });
-              }
-            });
-          } else {
-            // No need to close connection as transaction will abort and be returned to the pool
-          }
-        });
+    pool.withTransaction(client -> client
+      .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
+      .execute()
+      .flatMap(res -> client
+        .query("INSERT INTO Users (first_name,last_name) VALUES ('Julien','Viet')")
+        .execute()
+        // Map to a message result
+        .map("Users inserted"))
+    ).onComplete(ar -> {
+      // The connection was automatically return to the pool
+      if (ar.succeeded()) {
+        // Transaction was committed
+        String message = ar.result();
+        System.out.println("Transaction succeeded: " + message);
+      } else {
+        // Transaction was rolled back
+        System.out.println("Transaction failed " + ar.cause().getMessage());
       }
     });
   }
 
   public void usingCursors01(SqlConnection connection) {
-    connection.prepare("SELECT * FROM users WHERE first_name LIKE $1", ar1 -> {
-      if (ar1.succeeded()) {
-        PreparedStatement pq = ar1.result();
+    connection.prepare("SELECT * FROM users WHERE first_name LIKE $1", ar0 -> {
+      if (ar0.succeeded()) {
+        PreparedStatement pq = ar0.result();
 
         // Cursors require to run within a transaction
-        Transaction tx = connection.begin();
+        connection.begin(ar1 -> {
+          if (ar1.succeeded()) {
+            Transaction tx = ar1.result();
 
-        // Create a cursor
-        Cursor cursor = pq.cursor(Tuple.of("julien"));
+            // Create a cursor
+            Cursor cursor = pq.cursor(Tuple.of("julien"));
 
-        // Read 50 rows
-        cursor.read(50, ar2 -> {
-          if (ar2.succeeded()) {
-            RowSet<Row> rows = ar2.result();
-
-            // Check for more ?
-            if (cursor.hasMore()) {
-              // Repeat the process...
-            } else {
-              // No more rows - commit the transaction
-              tx.commit();
-            }
+            // Read 50 rows
+            cursor.read(50, ar2 -> {
+              if (ar2.succeeded()) {
+                RowSet<Row> rows = ar2.result();
+
+                // Check for more ?
+                if (cursor.hasMore()) {
+                  // Repeat the process...
+                } else {
+                  // No more rows - commit the transaction
+                  tx.commit();
+                }
+              }
+            });
           }
         });
       }
@@ -325,26 +324,30 @@ public void usingCursors02(Cursor cursor) {
   }
 
   public void usingCursors03(SqlConnection connection) {
-    connection.prepare("SELECT * FROM users WHERE first_name LIKE $1", ar1 -> {
-      if (ar1.succeeded()) {
-        PreparedStatement pq = ar1.result();
+    connection.prepare("SELECT * FROM users WHERE first_name LIKE $1", ar0 -> {
+      if (ar0.succeeded()) {
+        PreparedStatement pq = ar0.result();
 
         // Streams require to run within a transaction
-        Transaction tx = connection.begin();
+        connection.begin(ar1 -> {
+          if (ar1.succeeded()) {
+            Transaction tx = ar1.result();
 
-        // Fetch 50 rows at a time
-        RowStream<Row> stream = pq.createStream(50, Tuple.of("julien"));
+            // Fetch 50 rows at a time
+            RowStream<Row> stream = pq.createStream(50, Tuple.of("julien"));
 
-        // Use the stream
-        stream.exceptionHandler(err -> {
-          System.out.println("Error: " + err.getMessage());
-        });
-        stream.endHandler(v -> {
-          tx.commit();
-          System.out.println("End of stream");
-        });
-        stream.handler(row -> {
-          System.out.println("User: " + row.getString("last_name"));
+            // Use the stream
+            stream.exceptionHandler(err -> {
+              System.out.println("Error: " + err.getMessage());
+            });
+            stream.endHandler(v -> {
+              tx.commit();
+              System.out.println("End of stream");
+            });
+            stream.handler(row -> {
+              System.out.println("User: " + row.getString("last_name"));
+            });
+          }
         });
       }
     });

vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2SocketConnection.java

@@ -40,9 +40,9 @@ public class DB2SocketConnection extends SocketConnectionBase {
     private DB2Codec codec;
     private Handler<Void> closeHandler;
 
-    public DB2SocketConnection(NetSocketInternal socket, 
+    public DB2SocketConnection(NetSocketInternal socket,
     		boolean cachePreparedStatements,
-            int preparedStatementCacheSize, 
+            int preparedStatementCacheSize,
             int preparedStatementCacheSqlLimit,
             int pipeliningLimit,
             ContextInternal context) {
@@ -65,37 +65,37 @@ public void init() {
         pipeline.addBefore("handler", "codec", codec);
         super.init();
     }
-    
+
     @Override
     protected <R> void doSchedule(CommandBase<R> cmd, Handler<AsyncResult<R>> handler) {
       if (cmd instanceof TxCommand) {
-        TxCommand txCmd = (TxCommand) cmd;
-        if (TxCommand.BEGIN.sql.equals(txCmd.sql)) {
+        TxCommand<R> txCmd = (TxCommand<R>) cmd;
+        if (txCmd.kind == TxCommand.Kind.BEGIN) {
           // DB2 always implicitly starts a transaction with each query, and does
           // not support the 'BEGIN' keyword. Instead we can no-op BEGIN commands
           cmd.handler = handler;
-          cmd.complete(CommandResponse.success((R) null).toAsyncResult());
+          cmd.complete(CommandResponse.success(txCmd.result).toAsyncResult());
         } else {
           SimpleQueryCommand<Void> cmd2 = new SimpleQueryCommand<>(
-              txCmd.sql,
+              txCmd.kind.sql,
               false,
               false,
               QueryCommandBase.NULL_COLLECTOR,
               QueryResultHandler.NOOP_HANDLER);
-            super.doSchedule(cmd2, ar -> handler.handle(ar.mapEmpty()));
-          
+            super.doSchedule(cmd2, ar -> handler.handle(ar.map(txCmd.result)));
+
         }
       } else {
         super.doSchedule(cmd, handler);
       }
     }
-    
+
     @Override
     public void handleClose(Throwable t) {
       super.handleClose(t);
       context().runOnContext(closeHandler);
     }
-    
+
     public DB2SocketConnection closeHandler(Handler<Void> handler) {
       closeHandler = handler;
       return this;

vertx-db2-client/src/test/java/io/vertx/db2client/tck/DB2TransactionTest.java

@@ -37,13 +37,8 @@ public class DB2TransactionTest extends TransactionTestBase {
   public static DB2Resource rule = DB2Resource.SHARED_INSTANCE;
 
   @Override
-  protected void initConnector() {
-    connector = handler -> {
-      if (pool == null) {
-        pool = DB2Pool.pool(vertx, new DB2ConnectOptions(rule.options()), new PoolOptions().setMaxSize(1));
-      }
-      pool.begin(handler);
-    };
+  protected Pool createPool() {
+    return DB2Pool.pool(vertx, new DB2ConnectOptions(rule.options()), new PoolOptions().setMaxSize(1));
   }
 
   @Override
@@ -54,18 +49,14 @@ protected Pool nonTxPool() {
   @Override
   protected void cleanTestTable(TestContext ctx) {
     // use DELETE FROM because DB2 does not support TRUNCATE TABLE
-    connector.accept(ctx.asyncAssertSuccess(conn -> {
-      conn.query("DELETE FROM mutable").execute(ctx.asyncAssertSuccess(result -> {
-        conn.close();
-      }));
-    }));
+    getPool().query("DELETE FROM mutable").execute(ctx.asyncAssertSuccess());
   }
 
   @Override
   protected String statement(String... parts) {
     return String.join("?", parts);
   }
-  
+
   @Test
   public void testDelayedCommit(TestContext ctx) {
     assumeFalse("DB2 on Z holds write locks on inserted columns with isolation level = 2", rule.isZOS());