package com.atlassian.sal.core.rdbms;
   
   import com.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 org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  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 static final Logger log = LoggerFactory.getLogger(DefaultTransactionalExecutor.class);
  
      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, new ConnectionCallback<A>() {
              @Override
              public A execute(final Connection connection) {
                  return 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
              final A result = callback.execute(wrappedConnection);
              try {
                  // no exception indicates success
                  connection.commit();
              } catch (final Throwable se) {
                  // failure to commit should be propagated, to halt any further processing
                  throw new RdbmsException("Unable to commit connection", se);
              }
              return result;
         } catch (final Throwable re) {
             try {
                 // any exception indicates failure
                 connection.rollback();
             } catch (final Throwable se) {
                 re.addSuppressed(se);
             }
             throw re;
         }
     }
 
     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);
         }
     }
 }