import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * @param emit
 * @param <R>
 * @return
 */
@NonNull
@CheckReturnValue
public static <R> OnCompleteListener<R> listener(@NonNull final MaybeEmitter<R> emit) {
    return new OnCompleteListener<R>() {
        @Override
        public void onComplete(@NonNull final Task<R> task) {
            if (!emit.isDisposed()) {
                if (task.isSuccessful()) {
                    R result = task.getResult();
                    if (result != null) {
                        emit.onSuccess(result);
                    }
                    emit.onComplete();
                } else {
                    Exception e = task.getException();
                    emit.onError(e != null ? e : new RuntimeException());
                }
            }
        }
    };
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
@Override
@NonNull
@CheckReturnValue
public ObservableSource<DataValue<T>> apply(@NonNull Observable<DataSnapshot> upstream) {
    return upstream.map(new Function<DataSnapshot, DataValue<T>>() {
        @Override
        public DataValue<T> apply(DataSnapshot dataSnapshot) throws Exception {
            DataValue<T> result;
            if (dataSnapshot.exists()) {
                result = DataValue.of(dataSnapshot.getValue(typeIndicator));
            } else {
                result = DataValue.empty();
            }
            return result;
        }
    });
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
@Override
@NonNull
@CheckReturnValue
public ObservableSource<DataValue<T>> apply(@NonNull Observable<DataSnapshot> upstream) {
    return upstream.map(new Function<DataSnapshot, DataValue<T>>() {
        @Override
        public DataValue<T> apply(DataSnapshot dataSnapshot) throws Exception {
            DataValue<T> result;
            if (dataSnapshot.exists()) {
                result = DataValue.of(dataSnapshot.getValue(clazz));
            } else {
                result = DataValue.empty();
            }
            return result;
        }
    });
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * @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());
            }
        }
    };
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * @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();
        }
    });
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * 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;
        }
    });
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
@CheckReturnValue
@NonNull
public static <R> Observable<R> observable(@NonNull final Task<R> task) {
    return Observable.create(emitter -> {
        task.continueWith(t -> {
            if (emitter.isDisposed()) return null;

            if (t.isCancelled()) {
                // NOTICE: doOnUnsubscribe(() -> Observable.just(query) in outside
                emitter.onComplete();
            } else if (t.isFaulted()) {
                Throwable error = t.getError();
                emitter.onError(error);
            } else {
                R r = t.getResult();
                if (r != null) emitter.onNext(r);
                emitter.onComplete();
            }
            return null;
        });
    });
    // TODO .doOnUnsubscribe(() -> task.setCancelled());
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * Please don't put the Task<Void> as parameter
 * @param task
 * @param <R>
 * @return
 */
@CheckReturnValue
@NonNull
public static <R> Single<R> single(@NonNull final Task<R> task) {
    return Single.create(emitter -> {
        task.continueWith(t -> {
            if (emitter.isDisposed()) return null;

            if (t.isCancelled()) {
                emitter.onError(new RuntimeException("Cancelled task"));
            } else if (t.isFaulted()) {
                Throwable error = t.getError();
                emitter.onError(error);
            } else {
                R r = t.getResult();
                emitter.onSuccess(r);
            }
            return null;
        });
    });
    // TODO .doOnUnsubscribe(() -> task.setCancelled());
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 *  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());
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * 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);
    }
  };
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
@NonNull @CheckReturnValue @SuppressWarnings("checkstyle:overloadmethodsdeclarationorder") Observable<Permission> request(final Observable<?> trigger, @NonNull final String... permissions) {
  return Observable.merge(trigger, pending(permissions))
      .flatMap(new Function<Object, Observable<Permission>>() {
        @Override @NonNull @CheckReturnValue public Observable<Permission> apply(final Object o) throws Exception { // NOPMD
          return requestOnM(permissions);
        }
      });
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
@NonNull @CheckReturnValue private Observable<?> pending(@NonNull final String... permissions) {
  for (final String p : permissions) {
    if (!currentPermissionRequests.containsKey(p)) {
      return Observable.empty();
    }
  }

  return Observable.just(TRIGGER);
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
@NonNull @CheckReturnValue @TargetApi(M) Observable<Permission> requestOnM(@NonNull final String... permissions) {
  final List<Observable<Permission>> list = new ArrayList<>(permissions.length);
  final List<String> unrequestedPermissions = new ArrayList<>();

  // In case of multiple permissions, we create an observable for each of them.
  // At the end, the observables are combined to have a unique response.

  for (final String permission : permissions) {
    if (isGranted(permission)) {
      list.add(Observable.just(Permission.granted(permission)));
    } else if (isRevokedByPolicy(permission)) {
      list.add(Observable.just(Permission.revokedByPolicy(permission)));
    } else {
      PublishSubject<Permission> subject = currentPermissionRequests.get(permission);

      // Create a new subject if not exists
      if (subject == null) {
        unrequestedPermissions.add(permission);
        subject = PublishSubject.create();
        currentPermissionRequests.put(permission, subject);
      }

      list.add(subject);
    }
  }

  if (!unrequestedPermissions.isEmpty()) {
    final String[] permissionsToRequest = unrequestedPermissions.toArray(new String[0]);
    startShadowActivity(permissionsToRequest);
  }

  return Observable.concat(Observable.fromIterable(list));
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * @param callable
 * @param <R>
 * @return
 */
@CheckReturnValue
@NonNull
public static <R> Single<R> single(@NonNull final Callable<Task<R>> callable) {
    return Single.fromCallable(callable).flatMap(new Function<Task<R>,
            SingleSource<? extends R>>() {
        @Override
        public SingleSource<? extends R> apply(Task<R> task) throws Exception {
            return single(task);
        }
    });
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * @param task
 * @param <R>
 * @return
 */
@CheckReturnValue
@NonNull
public static <R> Single<R> single(@NonNull final Task<R> task) {
    return Single.create(new SingleOnSubscribe<R>() {
        @Override
        public void subscribe(@NonNull final SingleEmitter<R> emit) throws Exception {
            task.addOnCompleteListener(listener(emit));
        }
    });
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * @param callable
 * @param <R>
 * @return
 */
@CheckReturnValue
@NonNull
public static <R> Completable completes(@NonNull final Callable<Task<R>> callable) {
    return Single.fromCallable(callable).flatMapCompletable(
            new Function<Task<R>, Completable>() {
        @Override
        public Completable apply(Task<R> task) throws Exception {
            return completes(task);
        }
    });
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * @param callable
 * @param <R>
 * @return
 */
@CheckReturnValue
@NonNull
public static <R> Maybe<R> maybe(@NonNull final Callable<Task<R>> callable) {
    return Single.fromCallable(callable).flatMapMaybe(
            new Function<Task<R>, MaybeSource<? extends R>>() {
        @Override
        public MaybeSource<? extends R> apply(Task<R> task) throws Exception {
            return maybe(task);
        }
    });
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
/**
 * Auto close client
 * mqttConnectOptions.userName = it }$
 * mqttConnectOptions.password = it.toCharArray() }
 * @param client
 * @param topic
 * @return
 */
@NonNull
@CheckReturnValue
public static Observable<MqttMessage> remessage(@NonNull final MqttAndroidClient client,
                                                @NonNull final String topic) {
    final Observable<MqttMessage> msgObs =
            Observable.create(new ObservableOnSubscribe<MqttMessage>() {
        public void subscribe(
                @NonNull final ObservableEmitter<MqttMessage> emitter) throws Exception {
            client.subscribe(topic, 0, new IMqttMessageListener() {
                @Override
                public void messageArrived(
                        String topic2, @NonNull final MqttMessage message) throws Exception {
                    if (!emitter.isDisposed()) {
                        emitter.onNext(message);
                    }
                }
            });
        }
    });

    if (client.isConnected()) {
        return msgObs;
    } else {
        return reconnect(client).flatMapObservable(
                new Function<IMqttToken, ObservableSource<MqttMessage>>() {
            @Override
            public ObservableSource<MqttMessage> apply(IMqttToken token) throws Exception {
                return msgObs;
            }
        });
    }
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
@NonNull
@CheckReturnValue
public static Maybe<IMqttToken> connect(
        @NonNull final MqttAndroidClient client,
        @NonNull final DisconnectedBufferOptions disconnectedBufferOptions) {
    return connect(client, new MqttConnectOptions(), disconnectedBufferOptions);
}
 

import io.reactivex.annotations.CheckReturnValue; //导入依赖的package包/类
@NonNull
@CheckReturnValue
public static DisconnectedBufferOptions defaultDisconnectedBufferOptions() {
    final DisconnectedBufferOptions bufferOptions = new DisconnectedBufferOptions();
    bufferOptions.setBufferEnabled(true);
    bufferOptions.setBufferSize(100);
    bufferOptions.setPersistBuffer(false);
    bufferOptions.setDeleteOldestMessages(false);
    return bufferOptions;
}