Transaction States Should Be Checked In Queue (#189)

Motivation:
Currently, `MySqlConnection` checks state between the `Mono.defer` is
subscribed and `Exchangeable` is executed. It may cause undefined
behavior.

Modification:
Checks transaction state when request queue executes task.

Result:
Resolves #183

Author: jchrys

src/main/java/io/asyncer/r2dbc/mysql/ConnectionState.java

@@ -30,6 +30,13 @@ interface ConnectionState {
      */
     void setIsolationLevel(IsolationLevel level);
 
+    /**
+     * Reutrns session lock wait timeout.
+     *
+     * @return Session lock wait timeout.
+     */
+    long getSessionLockWaitTimeout();
+
     /**
      * Sets current lock wait timeout.
      *

src/main/java/io/asyncer/r2dbc/mysql/MySqlConnection.java

@@ -209,7 +209,7 @@ public Mono<Void> close() {
     @Override
     public Mono<Void> commitTransaction() {
         return Mono.defer(() -> {
-            return QueryFlow.doneTransaction(client, this, true, lockWaitTimeout, batchSupported);
+            return QueryFlow.doneTransaction(client, this, true, batchSupported);
         });
     }
 
@@ -223,19 +223,7 @@ public MySqlBatch createBatch() {
     @Override
     public Mono<Void> createSavepoint(String name) {
         requireValidName(name, "Savepoint name must not be empty and not contain backticks");
-
-        String sql = String.format("SAVEPOINT `%s`", name);
-
-        return Mono.defer(() -> {
-            if (isInTransaction()) {
-                return QueryFlow.executeVoid(client, sql);
-            } else if (batchSupported) {
-                // If connection does not in transaction, then starts transaction.
-                return QueryFlow.executeVoid(client, "BEGIN;" + sql);
-            }
-
-            return QueryFlow.executeVoid(client, "BEGIN", sql);
-        });
+        return QueryFlow.createSavepoint(client, this, name, batchSupported);
     }
 
     @Override
@@ -286,7 +274,7 @@ public Mono<Void> releaseSavepoint(String name) {
     @Override
     public Mono<Void> rollbackTransaction() {
         return Mono.defer(() -> {
-            return QueryFlow.doneTransaction(client, this, false, lockWaitTimeout, batchSupported);
+            return QueryFlow.doneTransaction(client, this, false, batchSupported);
         });
     }
 
@@ -371,6 +359,11 @@ public void setIsolationLevel(IsolationLevel level) {
         this.currentLevel = level;
     }
 
+    @Override
+    public long getSessionLockWaitTimeout() {
+        return lockWaitTimeout;
+    }
+
     @Override
     public void setCurrentLockWaitTimeout(long timeoutSeconds) {
         this.currentLockWaitTimeout = timeoutSeconds;