package com.atlassian.sal.core.rdbms;
   
   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 java.sql.Connection;
  import java.sql.SQLException;
  import javax.annotation.Nonnull;
  
  /**
   * 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);
              }
          });
      }
  
      @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);
         }
     }
 }