package com.atlassian.sal.core.rdbms;
   
   import com.atlassian.sal.api.rdbms.ConnectionCallback;
   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
          final WrappedConnection wrappedConnection = new WrappedConnection(connection);
          try
          {
              // execute the user's callback
              final A result = callback.execute(wrappedConnection);
              try
              {
                  // no exception indicates success
                  connection.commit();
             }
             catch (final SQLException se)
             {
                 // failure to commit should be propagated, to halt any further processing
                 throw new RuntimeException("Unable to commit connection", se);
             }
             return result;
         }
         catch (final RuntimeException re)
         {
             try
             {
                 // any exception indicates failure
                 connection.rollback();
             }
             catch (SQLException se)
             {
                 // swallow this because we want to rethrow the actual exception
                 log.error("Unable to rollback connection: " + se.getMessage());
             }
             throw re;
         }
         finally
         {
             // more protection against the user retaining the connection for later use
             wrappedConnection.expire();
         }
     }
 
     private void assertAutoCommitFalse(final Connection connection)
     {
         try
         {
             if (connection.getAutoCommit())
             {
                 throw new IllegalStateException("com.atlassian.sal.spi.HostConnectionAccessor returned connection with autocommit set");
             }
         }
         catch (SQLException e)
         {
             throw new RuntimeException("unable to invoke java.sql.Connection#getAutoCommit", e);
         }
     }
 }