/**
 * @param user
 * @param forceRefresh
 * @return
 */
@CheckReturnValue
@NonNull
public static Single<String> getToken(@NonNull final FirebaseUser user,
                                      final boolean forceRefresh) {
    return RxTask.single(new Callable<Task<GetTokenResult>>() {
        @Override
        public Task<GetTokenResult> call() throws Exception {
            return user.getToken(forceRefresh);
        }
    })
    .map(new Function<GetTokenResult, String>() {
        @Override
        public String apply(@NonNull GetTokenResult getTokenResult) throws Exception {
            return getTokenResult.getToken();
        }
    });
}
 

/**
 *  Limit 10000 by skip
 */
@NonNull
@CheckReturnValue
public static <R extends ParseObject> Observable<R> all(@NonNull final ParseQuery<R> query, int count) {
    final int limit = 1000; // limit limitation
    query.setSkip(0);
    query.setLimit(limit);
    Observable<R> find = find(query);
    for (int i = limit; i < count; i+= limit) {
        if (i >= 10000) break; // skip limitation
        query.setSkip(i);
        query.setLimit(limit);
        find.concatWith(find(query));
    }
    return find.distinct(o -> o.getObjectId());
}
 

/**
 * TODO: Should use Maybe instead of Single
 * TODO: flatten List
 *
 * @param instance
 * @param email
 * @return <emptyList> if providers is null
 */
@CheckReturnValue
@NonNull
public static Single<List<String>> fetchProvidersForEmail(
        @NonNull final FirebaseAuth instance, @NonNull final String email) {
    return RxTask.single(new Callable<Task<ProviderQueryResult>>() {
        @Override
        public Task<ProviderQueryResult> call() throws Exception {
            return instance.fetchProvidersForEmail(email);
        }
    }).map(new Function<ProviderQueryResult, List<String>>() {
        @Override
        public List<String> apply(@NonNull ProviderQueryResult providerQueryResult)
                throws Exception {
            List<String> providers = providerQueryResult.getProviders();
            if (null == providers) {
                providers = Collections.emptyList();
            }
            return providers;
        }
    });
}
 

/**
 * Converts an {@link ApolloQueryWatcher} to an asynchronous Observable.
 *
 * @param watcher the ApolloQueryWatcher to convert.
 * @param <T>     the value type
 * @return the converted Observable
 * @throws NullPointerException if watcher == null
 */
@NotNull
@CheckReturnValue
public static <T> Observable<Response<T>> from(@NotNull final ApolloQueryWatcher<T> watcher) {
  checkNotNull(watcher, "watcher == null");
  return Observable.create(new ObservableOnSubscribe<Response<T>>() {
    @Override public void subscribe(final ObservableEmitter<Response<T>> emitter) throws Exception {
      cancelOnObservableDisposed(emitter, watcher);

      watcher.enqueueAndWatch(new ApolloCall.Callback<T>() {
        @Override public void onResponse(@NotNull Response<T> response) {
          if (!emitter.isDisposed()) {
            emitter.onNext(response);
          }
        }

        @Override public void onFailure(@NotNull ApolloException e) {
          Exceptions.throwIfFatal(e);
          if (!emitter.isDisposed()) {
            emitter.onError(e);
          }
        }
      });
    }
  });
}
 

/**
 * Converts an {@link ApolloPrefetch} to a synchronous Completable
 *
 * @param prefetch the ApolloPrefetch to convert
 * @return the converted Completable
 * @throws NullPointerException if prefetch == null
 */
@NotNull
@CheckReturnValue
public static Completable from(@NotNull final ApolloPrefetch prefetch) {
  checkNotNull(prefetch, "prefetch == null");

  return Completable.create(new CompletableOnSubscribe() {
    @Override public void subscribe(final CompletableEmitter emitter) {
      cancelOnCompletableDisposed(emitter, prefetch);
      prefetch.enqueue(new ApolloPrefetch.Callback() {
        @Override public void onSuccess() {
          if (!emitter.isDisposed()) {
            emitter.onComplete();
          }
        }

        @Override public void onFailure(@NotNull ApolloException e) {
          Exceptions.throwIfFatal(e);
          if (!emitter.isDisposed()) {
            emitter.onError(e);
          }
        }
      });
    }
  });
}
 

/**
 * Converts an {@link ApolloStoreOperation} to a Single.
 *
 * @param operation the ApolloStoreOperation to convert
 * @param <T>       the value type
 * @return the converted Single
 */
@NotNull
@CheckReturnValue
public static <T> Single<T> from(@NotNull final ApolloStoreOperation<T> operation) {
  checkNotNull(operation, "operation == null");
  return Single.create(new SingleOnSubscribe<T>() {
    @Override
    public void subscribe(final SingleEmitter<T> emitter) {
      operation.enqueue(new ApolloStoreOperation.Callback<T>() {
        @Override
        public void onSuccess(T result) {
          emitter.onSuccess(result);
        }

        @Override
        public void onFailure(Throwable t) {
          emitter.onError(t);
        }
      });
    }
  });
}
 

/**
 * @param emit
 * @return
 */
@NonNull
@CheckReturnValue
public static ValueEventListener listener(@NonNull final SingleEmitter<DataSnapshot> emit) {
    return new ValueEventListener() {
        @Override
        public void onDataChange(DataSnapshot dataSnapshot) {
            if (!emit.isDisposed()) {
                emit.onSuccess(dataSnapshot);
            }
        }

        @Override
        public void onCancelled(DatabaseError e) {
            if (!emit.isDisposed()) {
                emit.onError(e.toException());
            }
        }
    };
}
 

@CheckReturnValue
public UpdateBuilder where(int column, String value) {
    String whereClauseText = DataBaseInfo.getFieldNames()[tablePosition][column] + " = ?";
    whereClause.add(whereClauseText);
    whereArgs.add(value);
    return this;
}
 

@CheckReturnValue
public UpdateBuilder where(String colName, String value) {
    DataBaseInfo.checkColumnName(colName, tablePosition);
    String whereClauseText = colName + " = ?";
    whereClause.add(whereClauseText);
    whereArgs.add(value);
    return this;
}
 

/**
 * @see StorageReference#putStream(InputStream)
 */
@CheckReturnValue
@NonNull
public static Single<UploadTask.TaskSnapshot> putStream(
        @NonNull final StorageReference ref,
        @NonNull final InputStream inputStream) {
    return RxTask.single(new Callable<Task<UploadTask.TaskSnapshot>>() {
        @Override
        public Task<UploadTask.TaskSnapshot> call() throws Exception {
            return ref.putStream(inputStream);
        }
    });
}
 

/**
 * @see StorageReference#getDownloadUrl()
 */
@CheckReturnValue
@NonNull
public static Single<Uri> getDownloadUrl(@NonNull final StorageReference ref) {
    return RxTask.single(new Callable<Task<Uri>>() {
        @Override
        public Task<Uri> call() throws Exception {
            return ref.getDownloadUrl();
        }
    });
}
 

/**
 * @see StorageReference#getStream()
 */
@CheckReturnValue
@NonNull
public static Single<StreamDownloadTask.TaskSnapshot> getStream(
        @NonNull final StorageReference ref) {
    return RxTask.single(new Callable<Task<StreamDownloadTask.TaskSnapshot>>() {
        @Override
        public Task<StreamDownloadTask.TaskSnapshot> call() throws Exception {
            return ref.getStream();
        }
    });
}
 

/**
 * TODO: AuthToUserFunction class
 * @return
 */
@CheckReturnValue
@NonNull
public static Function<AuthResult, FirebaseUser> authToUserFunction() {
    return new Function<AuthResult, FirebaseUser>() {
        @Override
        public FirebaseUser apply(@NonNull AuthResult authResult) throws Exception {
            return authResult.getUser();
        }
    };
}
 

/**
 * @param query
 * @param typeIndicator
 * @param <T>
 * @return
 */
@NonNull
@CheckReturnValue
public static <T> Single<T> dataOf(
        @NonNull Query query, @NonNull GenericTypeIndicator<T> typeIndicator) {
    return data(query).compose(new SingleTransformerOfGenericTypeIndicator<T>(typeIndicator));
}
 

/**
 * @param instance
 * @param token
 * @return
 */
@CheckReturnValue
@NonNull
public static Single<FirebaseUser> signInWithCustomToken(
        @NonNull final FirebaseAuth instance, @NonNull final String token) {
    return RxTask.single(new Callable<Task<AuthResult>>() {
        @Override
        public Task<AuthResult> call() throws Exception {
            return instance.signInWithCustomToken(token);
        }
    }).map(authToUserFunction());
}
 

/**
 * @param ref
 * @param fireLocalEvents
 * @param function
 * @return
 * ref.runTransaction(handler, fireLocalEvents);
 */
@NonNull
@CheckReturnValue
public static Single<DataSnapshot> runTransaction(
        @NonNull final DatabaseReference ref,
        @NonNull final Function<MutableData, Transaction.Result> function,
        final boolean fireLocalEvents) {
    return RxValue.transaction(ref, function, fireLocalEvents);
}
 

@CheckReturnValue
public QueryBuilder where(String colName, String value) {
    if (value == null) value = "";
    DataBaseInfo.checkColumnName(colName, tablePosition);
    String whereClauseText = colName + " = ?";
    whereClause.add(whereClauseText);
    whereArgs.add(value);
    return this;
}
 

@CheckReturnValue
public QueryBuilder setOrderByColumnAsc(String colName) {
    DataBaseInfo.checkColumnName(colName, tablePosition);
    isAsc = true;
    this.orderByColName = colName;
    return this;
}
 

@CheckReturnValue
public QueryBuilder setOrderByColumnDesc(String colName) {
    DataBaseInfo.checkColumnName(colName, tablePosition);
    isAsc = false;
    this.orderByColName = colName;
    return this;
}
 

@CheckReturnValue
private Cursor execute() {

    //selection
    String clause = null;
    String[] args = null;
    if (whereArgs.size() > 0) {
        args = new String[whereClause.size()];
        StringBuilder clauseBuilder = new StringBuilder();
        for (int i = 0; i < whereClause.size(); i++) {
            clauseBuilder.append(whereClause.get(i)).append(" AND ");
            args[i] = whereArgs.get(i);
        }
        if (clauseBuilder.length() > 5) {
            clause = clauseBuilder.substring(0, clauseBuilder.length() - 5);
        }
    }

    //order by
    String orderByText;
    if (isAsc) {
        orderByText = orderByColName == null ? null : orderByColName + " ASC";
    } else {
        orderByText = orderByColName == null ? null : orderByColName + " DESC";
    }

    if (sqLiteDatabase.isOpen())
        return sqLiteDatabase.query(DataBaseInfo.getTableNames()[tablePosition], colNames, clause, args, groupColName, having, orderByText, limit);
    return null;
}
 

/**
 * 查询数据库，不能在主线程执行
 *
 * 执行callback后返回值
 */
@CheckReturnValue
public synchronized <T> T executeAsync(QuerySyncResultCallback<T> callBack) {
    ActionBuilder.checkThreadLocal();

    Cursor cursor = execute();
    T result = callBack.onQuery(cursor);
    //自动检查Cursor是否关闭
    if (cursor != null && !cursor.isClosed()) {
        cursor.close();
    }
    return result;
}
 

/**
 * @param ref
 * @param function
 * @param fireLocalEvents
 * @return
 */
@NonNull
@CheckReturnValue
public static Single<DataSnapshot> transaction(
        @NonNull final DatabaseReference ref,
        @NonNull final Function<MutableData, Transaction.Result> function,
        final boolean fireLocalEvents) {
    return Single.create(new SingleOnSubscribe<DataSnapshot>() {
        @Override
        public void subscribe(
                @NonNull final SingleEmitter<DataSnapshot> emit) throws Exception {
            ref.runTransaction(transaction(emit, function), fireLocalEvents);
        }
    });
}
 

/**
 * @see StorageReference#getMetadata()
 */
@CheckReturnValue
@NonNull
public static Single<StorageMetadata> getMetadata(@NonNull final StorageReference ref) {
    return RxTask.single(new Callable<Task<StorageMetadata>>() {
        @Override
        public Task<StorageMetadata> call() throws Exception {
            return ref.getMetadata();
        }
    });
}
 

/**
 * Map emitted items from the source observable into {@link Permission} objects for each
 * permission in parameters.
 * <p>
 * If one or several permissions have never been requested, invoke the related framework method
 * to ask the user if he allows the permissions.
 */
@NonNull @CheckReturnValue private <T> ObservableTransformer<T, Permission> ensureEach(@NonNull final String... permissions) {
  checkPermissions(permissions);

  return new ObservableTransformer<T, Permission>() {
    @Override @NonNull @CheckReturnValue public ObservableSource<Permission> apply(final Observable<T> o) {
      return request(o, permissions);
    }
  };
}
 

/**
 * @see StorageReference#getBytes(long)
 */
@CheckReturnValue
@NonNull
public static Single<byte[]> getBytes(@NonNull final StorageReference ref,
                                      final long maxDownloadSizeBytes) {
    return RxTask.single(new Callable<Task<byte[]>>() {
        @Override
        public Task<byte[]> call() throws Exception {
            return ref.getBytes(maxDownloadSizeBytes);
        }
    });
}
 

/**
 * @param ref
 * @return
 */
@NonNull
@CheckReturnValue
public static Completable removeValue(@NonNull final DatabaseReference ref) {
    return RxTask.completes(new Callable<Task<Void>>() {
        @Override
        public Task<Void> call() throws Exception {
            return ref.removeValue();
        }
    });
}
 

/**
 * @param query
 * @return
 */
@NonNull
@CheckReturnValue
public static Observable<DataSnapshot> changes(@NonNull final Query query) {
    return Observable.create(new ObservableOnSubscribe<DataSnapshot>() {
        @Override
        public void subscribe(
                @NonNull final ObservableEmitter<DataSnapshot> emit) throws Exception {
            final ValueEventListener listener = new ValueEventListener() {
                @Override
                public void onDataChange(DataSnapshot dataSnapshot) {
                    if (!emit.isDisposed()) {
                        emit.onNext(dataSnapshot);
                    }
                }

                @Override
                public void onCancelled(DatabaseError e) {
                    if (!emit.isDisposed()) {
                        emit.onError(e.toException());
                    }
                }
            };

            emit.setCancellable(new Cancellable() {
                @Override
                public void cancel() throws Exception {
                    query.removeEventListener(listener);
                }
            });

            query.addValueEventListener(listener);
        }
    });
}
 

/**
 * @see StorageReference#putFile(Uri)
 */
@CheckReturnValue
@NonNull
public static Single<UploadTask.TaskSnapshot> putFile(@NonNull final StorageReference ref,
                                                      @NonNull final Uri uri) {
    return RxTask.single(new Callable<Task<UploadTask.TaskSnapshot>>() {
        @Override
        public Task<UploadTask.TaskSnapshot> call() throws Exception {
            return ref.putFile(uri);
        }
    });
}
 

/**
 * @param ref
 * @param function
 * @param fireLocalEvents
 * @return
 */
@NonNull
@CheckReturnValue
public static Single<DataSnapshot> transaction(
        @NonNull final DatabaseReference ref,
        @NonNull final Function<MutableData, Transaction.Result> function,
        final boolean fireLocalEvents) {
    return Single.create(new SingleOnSubscribe<DataSnapshot>() {
        @Override
        public void subscribe(
                @NonNull final SingleEmitter<DataSnapshot> emit) throws Exception {
            ref.runTransaction(transaction(emit, function), fireLocalEvents);
        }
    });
}
 

/**
 * @param emitter
 * @param function
 * @return
 */
@NonNull
@CheckReturnValue
public static Transaction.Handler transaction(
        @NonNull final SingleEmitter<DataSnapshot> emitter,
        @NonNull final Function<MutableData, Transaction.Result> function) {
    return new Transaction.Handler() {
        @Override
        public Transaction.Result doTransaction(MutableData mutableData) {
            try {
                return function.apply(mutableData);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public void onComplete(@Nullable DatabaseError databaseError,
                               boolean committed,
                               @NonNull DataSnapshot dataSnapshot) {
            if (!emitter.isDisposed()) {
                if (null == databaseError) {
                    emitter.onSuccess(dataSnapshot);
                } else {
                    emitter.onError(databaseError.toException());
                }
            }
        }
    };
}