package com.atlassian.sal.core.rdbms;
   
   import io.atlassian.fugue.Option;
   import com.atlassian.sal.api.rdbms.ConnectionCallback;
   import com.atlassian.sal.api.rdbms.RdbmsException;
   import com.atlassian.sal.api.rdbms.TransactionalExecutor;
   import com.atlassian.sal.spi.HostConnectionAccessor;
   import com.google.common.annotations.VisibleForTesting;
   
  import javax.annotation.Nonnull;
  import java.sql.Connection;
  import java.sql.SQLException;
  
  /**
   * Default implementation that invokes {@link com.atlassian.sal.spi.HostConnectionAccessor}.
   * <p>
   * Created by {@link com.atlassian.sal.core.rdbms.DefaultTransactionalExecutorFactory}
   *
   * @since 3.0
   */
  public class DefaultTransactionalExecutor implements TransactionalExecutor {
  
      private final HostConnectionAccessor hostConnectionAccessor;
  
      @VisibleForTesting
      boolean readOnly;
  
      @VisibleForTesting
      boolean newTransaction;
  
      public DefaultTransactionalExecutor(@Nonnull final HostConnectionAccessor hostConnectionAccessor,
                                          final boolean readOnly, final boolean newTransaction) {
          this.hostConnectionAccessor = hostConnectionAccessor;
          this.readOnly = readOnly;
          this.newTransaction = newTransaction;
      }
  
      @Override
      public <A> A execute(@Nonnull final ConnectionCallback<A> callback) {
          return hostConnectionAccessor.execute(
                  readOnly,
                  newTransaction,
                  connection -> executeInternal(connection, callback));
      }
  
      @Nonnull
      @Override
      public Option<String> getSchemaName() {
          return hostConnectionAccessor.getSchemaName();
      }
  
      @Override
      @Nonnull
      public TransactionalExecutor readOnly() {
          readOnly = true;
          return this;
      }
  
      @Override
      @Nonnull
      public TransactionalExecutor readWrite() {
          readOnly = false;
          return this;
      }
  
      @Override
      @Nonnull
      public TransactionalExecutor newTransaction() {
          newTransaction = true;
          return this;
      }
  
      @Override
      @Nonnull
      public TransactionalExecutor existingTransaction() {
          newTransaction = false;
          return this;
      }
  
      @VisibleForTesting
      <A> A executeInternal(@Nonnull final Connection connection, @Nonnull final ConnectionCallback<A> callback) {
          assertAutoCommitFalse(connection);
  
          // give the user the restricted connection
          try (final WrappedConnection wrappedConnection = new WrappedConnection(connection)) {
              // execute the user's callback
              return callback.execute(wrappedConnection);
          }
      }
  
      private void assertAutoCommitFalse(final Connection connection) {
          try {
              if (connection.getAutoCommit()) {
                  throw new IllegalStateException("com.atlassian.sal.spi.HostConnectionAccessor returned connection with autocommit set");
              }
          } catch (final SQLException e) {
              throw new RdbmsException("unable to invoke java.sql.Connection#getAutoCommit", e);
          }
      }
 }