import io.reactivex.CompletableSource; //导入依赖的package包/类
@Override
public CompletableSource apply(Completable upstream) {
    Completable tObservable = upstream
            .observeOn(AndroidSchedulers.mainThread());
    if (provider == null) {
        return tObservable;
    }
    return tObservable.compose(RxLifecycleDelegate.<T>bindLifecycle(provider));
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
/**
 * TakeUntil n'existe pas sur un completable. Amb permet de la premère émission
 *
 * @param upstream
 * @return
 */
@Override
public CompletableSource apply(Completable upstream) {
    return Completable.ambArray(upstream, mObservable.flatMapCompletable(new Function<Object, CompletableSource>() {
        @Override
        public CompletableSource apply(@NonNull Object o) throws Exception {
            return Completable.error(new CancellationException());
        }
    }));
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
private Disposable loadImage(final Uri uri) {
  mSourceUri = uri;
  return new RxPermissions(getActivity()).request(Manifest.permission.WRITE_EXTERNAL_STORAGE)
      .filter(new Predicate<Boolean>() {
        @Override public boolean test(@io.reactivex.annotations.NonNull Boolean granted)
            throws Exception {
          return granted;
        }
      })
      .flatMapCompletable(new Function<Boolean, CompletableSource>() {
        @Override
        public CompletableSource apply(@io.reactivex.annotations.NonNull Boolean aBoolean)
            throws Exception {
          return mCropView.load(uri)
              .useThumbnail(true)
              .initialFrameRect(mFrameRect)
              .executeAsCompletable();
        }
      })
      .subscribeOn(Schedulers.newThread())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new Action() {
        @Override public void run() throws Exception {
        }
      }, new Consumer<Throwable>() {
        @Override public void accept(@NonNull Throwable throwable) throws Exception {
        }
      });
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
@Override
public CompletableSource apply(Completable upstream) {
    return Completable.ambArray(upstream, observable.flatMapCompletable(new Function<Object, Completable>() {
        @Override
        public Completable apply(Object ignore) throws Exception {
            return Completable.error(new CancellationException());
        }
    }));
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
public static Completable animateTogether(final CompletableSource... completables) {
    return Completable.mergeArray(completables);
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
/**
 * @param source              the consumable CompletableSource
 * @param compositeDisposable the composite disposable
 */
public AttachDisposableCompletable(CompletableSource source, CompositeDisposable compositeDisposable) {
    this.source = source;
    this.compositeDisposable = compositeDisposable;
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
@Override
public final CompletableSource source() {
    return source;
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
@Override
public CompletableSource apply(Completable upstream) {
    return Completable.ambArray(upstream, observable.flatMapCompletable(Functions.CANCEL_COMPLETABLE));
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
CompletableIfThen(BooleanSupplier condition, CompletableSource then,
                  CompletableSource orElse) {
    this.condition = condition;
    this.then = then;
    this.orElse = orElse;
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
@Override
public CompletableSource source() {
    return mSource;
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
@Override
public CompletableSource apply(Completable upstream) {
    return upstream.toObservable()
            .compose(new DelayReplayObservableTransformer<>(pauseLifecycle))
            .ignoreElements();
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
AutoDisposeCompletable(CompletableSource source, Maybe<?> scope) {
  this.source = source;
  this.scope = scope;
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
@Override
public CompletableSource apply(Completable upstream) {
    return observableCache.cacheCompletable(key, upstream);
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
@Exclude
public Completable evaluateDeck(final CardsInteractor cardsInteractor) {
    return Completable.defer(new Callable<CompletableSource>() {
        @Override
        public CompletableSource call() throws Exception {
            return new Completable() {
                @Override
                protected void subscribeActual(final CompletableObserver emitter) {
                    if (cards == null) {
                        cards = new HashMap<String, CardDetails>();
                    }

                    cardsInteractor.getCard(leaderId)
                            .subscribeOn(Schedulers.io())
                            .observeOn(Schedulers.io())
                            .subscribe(
                            new BaseSingleObserver<CardDetails>() {
                                @Override
                                public void onSuccess(CardDetails value) {
                                    leader = value;
                                    leader.setPatch(patch);
                                    if (cards.keySet().size() == cardCount.keySet().size()) {
                                        emitter.onComplete();
                                    }
                                }
                            });

                    if (cardCount.keySet().size() == 0) {
                        return;
                    }

                    ArrayList<String> cardIds = new ArrayList<>();
                    for (String cardId : cardCount.keySet()) {
                        cardIds.add(cardId);
                    }

                    cardsInteractor.getCards(null, cardIds)
                            .subscribeOn(Schedulers.io())
                            .observeOn(Schedulers.io())
                            .subscribe(
                            new BaseSingleObserver<CardListResult>() {
                                @Override
                                public void onSuccess(CardListResult result) {
                                    if (result instanceof CardListResult.Success) {
                                        for (CardDetails card : ((CardListResult.Success) result).getCards()) {
                                            cards.put(card.getIngameId(), card);
                                        }
                                        if (leader != null) {
                                            emitter.onComplete();
                                        }
                                    }
                                }
                            });
                }
            };
        }
    });
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
@Override
public CompletableSource apply(Completable upstream) {
    return ((Observable<T>) apply(upstream.<T>toObservable())).ignoreElements();
}
 

import io.reactivex.CompletableSource; //导入依赖的package包/类
/**
 * TakeUntil n'existe pas sur un completable. Amb permet de la premère émission
 * @param upstream
 * @return
 */
@Override
public CompletableSource apply(Completable upstream) {
    return upstream;
}