/*
 * Copyright (c) 2011-2018 Contributors to the Eclipse Foundation
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
 * which is available at https://www.apache.org/licenses/LICENSE-2.0.
 *
 * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
 */

package io.vertx.reactivex.ext.sql;

import io.reactivex.*;
import io.reactivex.exceptions.Exceptions;
import io.vertx.reactivex.ext.sql.impl.*;

import java.util.function.Function;

/**
 * Utilities for generating observables with a {@link SQLClient}.
 *
 * @author Thomas Segismont
 */
public class SQLClientHelper {

  /**
   * Creates a {@link FlowableTransformer} decorating a {@link Flowable} with transaction management for a given {@link SQLConnection}.
   * <p>
   * If the upstream {@link Flowable} completes (<em>onComplete</em>), the transaction is committed.
   * If the upstream {@link Flowable} emits an error (<em>onError</em>), the transaction is rollbacked.
   * <p>
   * Eventually, the given {@link SQLConnection} is put back in <em>autocommit</em> mode.
   *
   * @param sqlConnection the {@link SQLConnection} used for database operations and transaction management
   * @param <T> the type of the items emitted by the upstream {@link Flowable}
   * @return a {@link FlowableTransformer} decorating a {@link Flowable} with transaction management
   */
  public static <T> FlowableTransformer<T, T> txFlowableTransformer(SQLConnection sqlConnection) {
    return new InTransactionFlowable<>(sqlConnection);
  }

  /**
   * Generates a {@link Flowable} from {@link SQLConnection} operations executed inside a transaction.
   *
   * @param client the {@link SQLClient}
   * @param sourceSupplier a user-provided function returning a {@link Flowable} generated by interacting with the given {@link SQLConnection}
   * @param <T> the type of the items emitted by the {@link Flowable}
   * @return a {@link Flowable} generated from {@link SQLConnection} operations executed inside a transaction
   */
  public static <T> Flowable<T> inTransactionFlowable(SQLClient client, Function<SQLConnection, Flowable<T>> sourceSupplier) {
    return usingConnectionFlowable(client, conn -> sourceSupplier.apply(conn).compose(txFlowableTransformer(conn)));
  }

  /**
   * Creates a {@link ObservableTransformer} decorating an {@link Observable} with transaction management for a given {@link SQLConnection}.
   * <p>
   * If the upstream {@link Observable} completes (<em>onComplete</em>), the transaction is committed.
   * If the upstream {@link Observable} emits an error (<em>onError</em>), the transaction is rollbacked.
   * <p>
   * Eventually, the given {@link SQLConnection} is put back in <em>autocommit</em> mode.
   *
   * @param sqlConnection the {@link SQLConnection} used for database operations and transaction management
   * @param <T> the type of the items emitted by the upstream {@link Observable}
   * @return a {@link ObservableTransformer} decorating an {@link Observable} with transaction management
   */
  public static <T> ObservableTransformer<T, T> txObservableTransformer(SQLConnection sqlConnection) {
    return new InTransactionObservable<>(sqlConnection);
  }

  /**
   * Generates a {@link Observable} from {@link SQLConnection} operations executed inside a transaction.
   *
   * @param client the {@link SQLClient}
   * @param sourceSupplier a user-provided function returning a {@link Observable} generated by interacting with the given {@link SQLConnection}
   * @param <T> the type of the items emitted by the {@link Observable}
   * @return an {@link Observable} generated from {@link SQLConnection} operations executed inside a transaction
   */
  public static <T> Observable<T> inTransactionObservable(SQLClient client, Function<SQLConnection, Observable<T>> sourceSupplier) {
    return usingConnectionObservable(client, conn -> sourceSupplier.apply(conn).compose(txObservableTransformer(conn)));
  }

  /**
   * Creates a {@link SingleTransformer} decorating a {@link Single} with transaction management for a given {@link SQLConnection}.
   * <p>
   * If the upstream {@link Single} emits a value (<em>onSuccess</em>), the transaction is committed.
   * If the upstream {@link Single} emits an error (<em>onError</em>), the transaction is rollbacked.
   * <p>
   * Eventually, the given {@link SQLConnection} is put back in <em>autocommit</em> mode.
   *
   * @param sqlConnection the {@link SQLConnection} used for database operations and transaction management
   * @param <T> the type of the item emitted by the upstream {@link Single}
   * @return a {@link SingleTransformer} decorating a {@link Single} with transaction management
   */
  public static <T> SingleTransformer<T, T> txSingleTransformer(SQLConnection sqlConnection) {
    return new InTransactionSingle<>(sqlConnection);
  }

  /**
   * Generates a {@link Single} from {@link SQLConnection} operations executed inside a transaction.
   *
   * @param client the {@link SQLClient}
   * @param sourceSupplier a user-provided function returning a {@link Single} generated by interacting with the given {@link SQLConnection}
   * @param <T> the type of the item emitted by the {@link Single}
   * @return a {@link Single} generated from {@link SQLConnection} operations executed inside a transaction
   */
  public static <T> Single<T> inTransactionSingle(SQLClient client, Function<SQLConnection, Single<T>> sourceSupplier) {
    return usingConnectionSingle(client, conn -> sourceSupplier.apply(conn).compose(txSingleTransformer(conn)));
  }

  /**
   * Creates a {@link MaybeTransformer} decorating a {@link Maybe} with transaction management for a given {@link SQLConnection}.
   * <p>
   * If the upstream {@link Maybe} emits a value (<em>onSuccess</em>) or completes (<em>onComplete</em>), the transaction is committed.
   * If the upstream {@link Maybe} emits an error (<em>onError</em>), the transaction is rollbacked.
   * <p>
   * Eventually, the given {@link SQLConnection} is put back in <em>autocommit</em> mode.
   *
   * @param sqlConnection the {@link SQLConnection} used for database operations and transaction management
   * @param <T> the type of the item emitted by the upstream {@link Maybe}
   * @return a {@link MaybeTransformer} decorating a {@link Maybe} with transaction management
   */
  public static <T> MaybeTransformer<T, T> txMaybeTransformer(SQLConnection sqlConnection) {
    return new InTransactionMaybe<>(sqlConnection);
  }

  /**
   * Generates a {@link Maybe} from {@link SQLConnection} operations executed inside a transaction.
   *
   * @param client the {@link SQLClient}
   * @param sourceSupplier a user-provided function returning a {@link Maybe} generated by interacting with the given {@link SQLConnection}
   * @param <T> the type of the item emitted by the {@link Maybe}
   * @return a {@link Maybe} generated from {@link SQLConnection} operations executed inside a transaction
   */
  public static <T> Maybe<T> inTransactionMaybe(SQLClient client, Function<SQLConnection, Maybe<T>> sourceSupplier) {
    return usingConnectionMaybe(client, conn -> sourceSupplier.apply(conn).compose(txMaybeTransformer(conn)));
  }

  /**
   * Creates a {@link CompletableTransformer} decorating a {@link Completable} with transaction management for a given {@link SQLConnection}.
   * <p>
   * If the upstream {@link Completable} completes (<em>onComplete</em>), the transaction is committed.
   * If the upstream {@link Completable} emits an error (<em>onError</em>), the transaction is rollbacked.
   * <p>
   * Eventually, the given {@link SQLConnection} is put back in <em>autocommit</em> mode.
   *
   * @param sqlConnection the {@link SQLConnection} used for database operations and transaction management
   * @return a {@link CompletableTransformer} decorating a {@link Completable} with transaction management
   */
  public static CompletableTransformer txCompletableTransformer(SQLConnection sqlConnection) {
    return new InTransactionCompletable(sqlConnection);
  }

  /**
   * Generates a {@link Completable} from {@link SQLConnection} operations executed inside a transaction.
   *
   * @param client the {@link SQLClient}
   * @param sourceSupplier a user-provided function returning a {@link Completable} generated by interacting with the given {@link SQLConnection}
   * @return a {@link Completable} generated from {@link SQLConnection} operations executed inside a transaction
   */
  public static Completable inTransactionCompletable(SQLClient client, Function<SQLConnection, Completable> sourceSupplier) {
    return usingConnectionCompletable(client, conn -> sourceSupplier.apply(conn).compose(txCompletableTransformer(conn)));
  }

  /**
   * Generates a {@link Flowable} from {@link SQLConnection} operations.
   *
   * @param client the {@link SQLClient}
   * @param sourceSupplier a user-provided function returning a {@link Flowable} generated by interacting with the given {@link SQLConnection}
   * @param <T> the type of the items emitted by the {@link Flowable}
   * @return a {@link Flowable} generated from {@link SQLConnection} operations
   */
  public static <T> Flowable<T> usingConnectionFlowable(SQLClient client, Function<SQLConnection, Flowable<T>> sourceSupplier) {
    return client.rxGetConnection().flatMapPublisher(conn -> {
      try {
        return sourceSupplier.apply(conn).doFinally(conn::close);
      } catch (Throwable t) {
        Exceptions.throwIfFatal(t);
        conn.close();
        return Flowable.error(t);
      }
    });
  }

  /**
   * Generates a {@link Observable} from {@link SQLConnection} operations.
   *
   * @param client the {@link SQLClient}
   * @param sourceSupplier a user-provided function returning a {@link Observable} generated by interacting with the given {@link SQLConnection}
   * @param <T> the type of the items emitted by the {@link Observable}
   * @return an {@link Observable} generated from {@link SQLConnection} operations
   */
  public static <T> Observable<T> usingConnectionObservable(SQLClient client, Function<SQLConnection, Observable<T>> sourceSupplier) {
    return client.rxGetConnection().flatMapObservable(conn -> {
      try {
        return sourceSupplier.apply(conn).doFinally(conn::close);
      } catch (Throwable t) {
        Exceptions.throwIfFatal(t);
        conn.close();
        return Observable.error(t);
      }
    });
  }

  /**
   * Generates a {@link Single} from {@link SQLConnection} operations.
   *
   * @param client the {@link SQLClient}
   * @param sourceSupplier a user-provided function returning a {@link Single} generated by interacting with the given {@link SQLConnection}
   * @param <T> the type of the item emitted by the {@link Single}
   * @return a {@link Single} generated from {@link SQLConnection} operations
   */
  public static <T> Single<T> usingConnectionSingle(SQLClient client, Function<SQLConnection, Single<T>> sourceSupplier) {
    return client.rxGetConnection().flatMap(conn -> {
      try {
        return sourceSupplier.apply(conn).doFinally(conn::close);
      } catch (Throwable t) {
        Exceptions.throwIfFatal(t);
        conn.close();
        return Single.error(t);
      }
    });
  }

  /**
   * Generates a {@link Maybe} from {@link SQLConnection} operations.
   *
   * @param client the {@link SQLClient}
   * @param sourceSupplier a user-provided function returning a {@link Maybe} generated by interacting with the given {@link SQLConnection}
   * @param <T> the type of the item emitted by the {@link Maybe}
   * @return a {@link Maybe} generated from {@link SQLConnection} operations
   */
  public static <T> Maybe<T> usingConnectionMaybe(SQLClient client, Function<SQLConnection, Maybe<T>> sourceSupplier) {
    return client.rxGetConnection().flatMapMaybe(conn -> {
      try {
        return sourceSupplier.apply(conn).doFinally(conn::close);
      } catch (Throwable t) {
        Exceptions.throwIfFatal(t);
        conn.close();
        return Maybe.error(t);
      }
    });
  }

  /**
   * Generates a {@link Completable} from {@link SQLConnection} operations.
   *
   * @param client the {@link SQLClient}
   * @param sourceSupplier a user-provided function returning a {@link Completable} generated by interacting with the given {@link SQLConnection}
   * @return a {@link Completable} generated from {@link SQLConnection} operations
   */
  public static Completable usingConnectionCompletable(SQLClient client, Function<SQLConnection, Completable> sourceSupplier) {
    return client.rxGetConnection().flatMapCompletable(conn -> {
      try {
        return sourceSupplier.apply(conn).doFinally(conn::close);
      } catch (Throwable t) {
        Exceptions.throwIfFatal(t);
        conn.close();
        return Completable.error(t);
      }
    });
  }

  private SQLClientHelper() {
    // Utility
  }
}
