心动空间 v1.0.0版本的 MD5 值为:de8941a869ff7ed1429262fe5aa87293
以下内容为反编译后的 CompletableLife.java 源代码,内容仅作参考
package com.rxjava.rxlife;
import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.core.Completable;
import io.reactivex.rxjava3.core.CompletableObserver;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.exceptions.Exceptions;
import io.reactivex.rxjava3.functions.Action;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.internal.functions.Functions;
import io.reactivex.rxjava3.internal.observers.CallbackCompletableObserver;
import io.reactivex.rxjava3.internal.observers.EmptyCompletableObserver;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
public class CompletableLife extends RxSource<CompletableObserver> {
private final Completable upStream;
@Override
public Object subscribeWith(Object obj) {
return super.subscribeWith(obj);
}
public CompletableLife(Completable completable, Scope scope, boolean z) {
super(scope, z);
this.upStream = completable;
}
@Override
public final Disposable subscribe() {
EmptyCompletableObserver emptyCompletableObserver = new EmptyCompletableObserver();
subscribe((CompletableObserver) emptyCompletableObserver);
return emptyCompletableObserver;
}
public final Disposable subscribe(Action action) {
return subscribe(action, Functions.ON_ERROR_MISSING);
}
public final Disposable subscribe(Action action, Consumer<? super Throwable> consumer) {
ObjectHelper.requireNonNull(consumer, "onError is null");
ObjectHelper.requireNonNull(action, "onComplete is null");
CallbackCompletableObserver callbackCompletableObserver = new CallbackCompletableObserver(consumer, action);
subscribe((CompletableObserver) callbackCompletableObserver);
return callbackCompletableObserver;
}
@Override
public final void subscribe(CompletableObserver completableObserver) {
ObjectHelper.requireNonNull(completableObserver, "observer is null");
try {
CompletableObserver onSubscribe = RxJavaPlugins.onSubscribe(this.upStream, completableObserver);
ObjectHelper.requireNonNull(onSubscribe, "The RxJavaPlugins.onSubscribe hook returned a null CompletableObserver. Please check the handler provided to RxJavaPlugins.setOnCompletableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
subscribeActual(onSubscribe);
} catch (NullPointerException e) {
throw e;
} catch (Throwable th) {
Exceptions.throwIfFatal(th);
RxJavaPlugins.onError(th);
NullPointerException nullPointerException = new NullPointerException("Actually not, but can't pass out an exception otherwise...");
nullPointerException.initCause(th);
throw nullPointerException;
}
}
private void subscribeActual(CompletableObserver completableObserver) {
Completable completable = this.upStream;
if (this.onMain) {
completable = completable.observeOn(AndroidSchedulers.mainThread());
}
completable.onTerminateDetach().subscribe(new LifeCompletableObserver(completableObserver, this.scope));
}
}