package org.davidmoten.rx.jdbc;
   
   import java.sql.Connection;
   import java.sql.ResultSet;
   import java.util.List;
   import java.util.concurrent.atomic.AtomicReference;
   
   import javax.annotation.Nonnull;
   
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import io.reactivex.Flowable;
  import io.reactivex.Single;
  
  public final class TransactedUpdateBuilder implements DependsOn<TransactedUpdateBuilder> {
  
      private static final Logger log = LoggerFactory.getLogger(TransactedUpdateBuilder.class);
  
      final UpdateBuilder updateBuilder;
      private final Database db;
      private boolean valuesOnly;
  
      TransactedUpdateBuilder(UpdateBuilder b, Database db) {
          this.updateBuilder = b;
          this.db = db;
      }
  
      public TransactedUpdateBuilder parameterStream(@Nonnull Flowable<?> values) {
          updateBuilder.parameterStream(values);
          return this;
      }
  
      public TransactedUpdateBuilder parameterListStream(@Nonnull Flowable<List<?>> valueLists) {
          updateBuilder.parameterListStream(valueLists);
          return this;
      }
  
      public TransactedUpdateBuilder parameters(@Nonnull List<?> values) {
          updateBuilder.parameters(values);
          return this;
      }
  
      public TransactedUpdateBuilder parameter(@Nonnull String name, Object value) {
          updateBuilder.parameter(name, value);
          return this;
      }
  
      public TransactedUpdateBuilder parameter(Object value) {
          return parameters(value);
      }
  
      public TransactedUpdateBuilder parameters(@Nonnull Object... values) {
          updateBuilder.parameters(values);
          return this;
      }
  
      @Override
      public TransactedUpdateBuilder dependsOn(@Nonnull Flowable<?> dependency) {
          updateBuilder.dependsOn(dependency);
          return this;
      }
  
      public TransactedUpdateBuilder batchSize(int batchSize) {
          updateBuilder.batchSize(batchSize);
          return this;
      }
  
      /**
       * Returns a builder used to specify how to process the generated keys
       * {@link ResultSet}. Not all jdbc drivers support this functionality and some
       * have limitations in their support (h2 for instance only returns the last
       * generated key when multiple inserts happen in the one statement).
       * 
       * @return a builder used to specify how to process the generated keys ResultSet
       */
      public TransactedReturnGeneratedKeysBuilder returnGeneratedKeys() {
          return new TransactedReturnGeneratedKeysBuilder(this, db);
      }
  
      public TransactedUpdateBuilder transactedValuesOnly() {
          this.valuesOnly = true;
          return this;
      }
  
      public TransactedUpdateBuilderValuesOnly valuesOnly() {
          return new TransactedUpdateBuilderValuesOnly(this, db);
      }
  
      public static final class TransactedUpdateBuilderValuesOnly {
          private final TransactedUpdateBuilder b;
          private final Database db;
  
          TransactedUpdateBuilderValuesOnly(TransactedUpdateBuilder b, Database db) {
              this.b = b;
              this.db = db;
          }
  
          // TODO add other methods e.g. parameter setting methods? Lots of
         // copy-and-paste not attractive here so may accept restricting
         // functionality once valuesOnly() called
 
         public Flowable<Integer> counts() {
             return createFlowable(b.updateBuilder, db) //
                     .flatMap(Tx.flattenToValuesOnly());
         }
     }
 
     public Flowable<Tx<Integer>> counts() {
         Flowable<Tx<Integer>> o = createFlowable(updateBuilder, db);
         if (valuesOnly) {
             return o.filter(tx -> tx.isValue());
         } else {
             return o;
         }
     }
 
     public Flowable<Integer> countsOnly() {
         return valuesOnly().counts();
     }
 
     @SuppressWarnings("unchecked")
     public Flowable<Tx<?>> tx() {
         return (Flowable<Tx<?>>) (Flowable<?>) createFlowable(updateBuilder, db) //
                 .filter(x -> x.isValue());
     }
 
     private static Flowable<Tx<Integer>> createFlowable(UpdateBuilder ub, Database db) {
         return Flowable.defer(() -> {
             log.debug("creating deferred flowable");
             AtomicReference<Connection> connection = new AtomicReference<Connection>();
             Single<Connection> con = ub.connections //
                     .map(c -> Util.toTransactedConnection(connection, c));
             TxImpl<?>[] t = new TxImpl[1];
             return ub.startWithDependency( //
                     Update.create(con, //
                             ub.parameterGroupsToFlowable(), //
                             ub.sql, //
                             ub.batchSize, //
                             false, //
                             ub.queryTimeoutSec) //
                             .flatMap(n -> Tx.toTx(n, connection.get(), db)) //
                             .doOnNext(tx -> {
                                 t[0] = ((TxImpl<Integer>) tx);
                             }) //
                             .doOnComplete(() -> {
                                 TxImpl<?> tx = t[0];
                                 if (tx.isComplete()) {
                                     tx.connection().commit();
                                 }
                                 Util.closeSilently(tx.connection());
                             }));
         });
     }
 
     public Flowable<List<Object>> parameterGroupsToFlowable() {
         return updateBuilder.parameterGroupsToFlowable();
     }
 }