// 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;
    if (isInReadonlyTransaction()) {
      closeReadOnlyTransaction();
    }

    return returnFuture;
  }

  /**
   * Commits the current transaction (if read/write) or closes it (if read-only).
   *
   * @return chainable {@link ApiFuture} for commit status.
   */
  ApiFuture<Timestamp> commitTransaction() {
    if (isInReadWriteTransaction()) {
      if (this.lastStep == null) {
        LOGGER.warn("Read/Write transaction committing without any statements.");
        return ApiFutures.immediateFuture(null);
      }
      return this.lastStep.commitAsync();

    } else if (isInReadonlyTransaction()) {
      closeReadOnlyTransaction();
    } else {
      LOGGER.warn("Commit called outside of an active transaction.");
    }
    return ApiFutures.immediateFuture(null);
  }

  private void closeReadOnlyTransaction() {
    this.readOnlyTransaction.close();
// 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;

    if (isInReadonlyTransaction()) {
      closeReadOnlyTransaction();
    }

    return returnFuture;
  }

  /**
   * Commits the current transaction (if read/write) or closes it (if read-only).
   *
   * @return chainable {@link ApiFuture} for commit status.
   */
  ApiFuture<Timestamp> commitTransaction() {
    if (isInReadWriteTransaction()) {
      if (this.lastStep == null) {
        LOGGER.warn("Read/Write transaction committing without any statements.");
        return ApiFutures.immediateFuture(null);
      }
      return this.lastStep.commitAsync();

    } else if (isInReadonlyTransaction()) {
      closeReadOnlyTransaction();
    } else {
      LOGGER.warn("Commit called outside of an active transaction.");
    }
    return ApiFutures.immediateFuture(null);
  }

  private void closeReadOnlyTransaction() {
    this.readOnlyTransaction.close();
    this.readOnlyTransaction = null;