// Path: cloud-spanner-r2dbc/src/main/java/com/google/cloud/spanner/r2dbc/v2/DatabaseClientReactiveAdapter.java
// public static final Executor REACTOR_EXECUTOR =
//     runnable -> Schedulers.parallel().schedule(runnable);
// 
// Path: cloud-spanner-r2dbc/src/main/java/com/google/cloud/spanner/r2dbc/TransactionInProgressException.java
// public class TransactionInProgressException extends R2dbcNonTransientException {
// 
//   public static final String MSG_READONLY =
//       "Cannot begin a new transaction because a readonly transaction is already in progress.";
//   public static final String MSG_READWRITE =
//       "Cannot begin a new transaction because a read/write transaction is already in progress.";
// 
//   public TransactionInProgressException(boolean isReadwrite) {
//     super(isReadwrite ? MSG_READWRITE : MSG_READONLY);
//   }
// 
// }


import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static com.google.cloud.spanner.r2dbc.v2.DatabaseClientReactiveAdapter.REACTOR_EXECUTOR;
import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutures;
import com.google.cloud.Timestamp;
import com.google.cloud.spanner.AsyncTransactionManager;
import com.google.cloud.spanner.AsyncTransactionManager.AsyncTransactionStep;
import com.google.cloud.spanner.AsyncTransactionManager.TransactionContextFuture;
import com.google.cloud.spanner.DatabaseClient;
import com.google.cloud.spanner.ReadContext;
import com.google.cloud.spanner.ReadOnlyTransaction;
import com.google.cloud.spanner.TimestampBound;
import com.google.cloud.spanner.TransactionContext;
import com.google.cloud.spanner.r2dbc.TransactionInProgressException;
   */
  ApiFuture<Void> rollbackTransaction() {

    if (isInReadWriteTransaction()) {
      if (this.lastStep == null) {
        LOGGER.warn("Read/Write transaction rolling back without any statements.");
      }
      return this.transactionManager.rollbackAsync();
    }
    LOGGER.warn("Rollback called outside of an active read/write transaction.");
    return ApiFutures.immediateFuture(null);
  }

  /**
   * Runs provided operation, managing the client library transactional future chaining.
   *
   * @param operation a function executing either streaming SQL or DML.
   *     The function accepts ReadContext for SELECT queries, and TransactionContext for DML.
   * @param <T> Type of object wrapped by the {@link ApiFuture} returned by the operation
   *
   * @return {@link ApiFuture} result of the provided operation
   */
  <T> ApiFuture<T> runInTransaction(Function<? super TransactionContext, ApiFuture<T>> operation) {

    // The first statement in a transaction has no input, hence Void input type.
    // The subsequent statements take the previous statements' return (affected row count)
    // as input.
    AsyncTransactionStep<? extends Object, T> updateStatementFuture =
        this.lastStep == null
            ? this.txnContextFuture.then(
// Path: cloud-spanner-r2dbc/src/main/java/com/google/cloud/spanner/r2dbc/v2/DatabaseClientReactiveAdapter.java
// public static final Executor REACTOR_EXECUTOR =
//     runnable -> Schedulers.parallel().schedule(runnable);
// 
// Path: cloud-spanner-r2dbc/src/main/java/com/google/cloud/spanner/r2dbc/TransactionInProgressException.java
// public class TransactionInProgressException extends R2dbcNonTransientException {
// 
//   public static final String MSG_READONLY =
//       "Cannot begin a new transaction because a readonly transaction is already in progress.";
//   public static final String MSG_READWRITE =
//       "Cannot begin a new transaction because a read/write transaction is already in progress.";
// 
//   public TransactionInProgressException(boolean isReadwrite) {
//     super(isReadwrite ? MSG_READWRITE : MSG_READONLY);
//   }
// 
// }


// Path: cloud-spanner-r2dbc/src/main/java/com/google/cloud/spanner/r2dbc/v2/DatabaseClientTransactionManager.java
import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static com.google.cloud.spanner.r2dbc.v2.DatabaseClientReactiveAdapter.REACTOR_EXECUTOR;
import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutures;
import com.google.cloud.Timestamp;
import com.google.cloud.spanner.AsyncTransactionManager;
import com.google.cloud.spanner.AsyncTransactionManager.AsyncTransactionStep;
import com.google.cloud.spanner.AsyncTransactionManager.TransactionContextFuture;
import com.google.cloud.spanner.DatabaseClient;
import com.google.cloud.spanner.ReadContext;
import com.google.cloud.spanner.ReadOnlyTransaction;
import com.google.cloud.spanner.TimestampBound;
import com.google.cloud.spanner.TransactionContext;
import com.google.cloud.spanner.r2dbc.TransactionInProgressException;

   */
  ApiFuture<Void> rollbackTransaction() {

    if (isInReadWriteTransaction()) {
      if (this.lastStep == null) {
        LOGGER.warn("Read/Write transaction rolling back without any statements.");
      }
      return this.transactionManager.rollbackAsync();
    }
    LOGGER.warn("Rollback called outside of an active read/write transaction.");
    return ApiFutures.immediateFuture(null);
  }

  /**
   * Runs provided operation, managing the client library transactional future chaining.
   *
   * @param operation a function executing either streaming SQL or DML.
   *     The function accepts ReadContext for SELECT queries, and TransactionContext for DML.
   * @param <T> Type of object wrapped by the {@link ApiFuture} returned by the operation
   *
   * @return {@link ApiFuture} result of the provided operation
   */
  <T> ApiFuture<T> runInTransaction(Function<? super TransactionContext, ApiFuture<T>> operation) {

    // The first statement in a transaction has no input, hence Void input type.
    // The subsequent statements take the previous statements' return (affected row count)
    // as input.
    AsyncTransactionStep<? extends Object, T> updateStatementFuture =
        this.lastStep == null
            ? this.txnContextFuture.then(
                (ctx, unusedVoid) -> operation.apply(ctx), REACTOR_EXECUTOR)