73 lines
2.3 KiB
Java
73 lines
2.3 KiB
Java
package retrofit2.adapter.rxjava2;
|
|
|
|
import io.reactivex.Observable;
|
|
import io.reactivex.Observer;
|
|
import io.reactivex.disposables.Disposable;
|
|
import io.reactivex.exceptions.CompositeException;
|
|
import io.reactivex.exceptions.Exceptions;
|
|
import io.reactivex.plugins.RxJavaPlugins;
|
|
import retrofit2.Response;
|
|
|
|
/* loaded from: classes2.dex */
|
|
final class BodyObservable<T> extends Observable<T> {
|
|
private final Observable<Response<T>> upstream;
|
|
|
|
private static class BodyObserver<R> implements Observer<Response<R>> {
|
|
private final Observer<? super R> observer;
|
|
private boolean terminated;
|
|
|
|
BodyObserver(Observer<? super R> observer) {
|
|
this.observer = observer;
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onComplete() {
|
|
if (this.terminated) {
|
|
return;
|
|
}
|
|
this.observer.onComplete();
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onError(Throwable th) {
|
|
if (!this.terminated) {
|
|
this.observer.onError(th);
|
|
return;
|
|
}
|
|
AssertionError assertionError = new AssertionError("This should never happen! Report as a bug with the full stacktrace.");
|
|
assertionError.initCause(th);
|
|
RxJavaPlugins.b(assertionError);
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onSubscribe(Disposable disposable) {
|
|
this.observer.onSubscribe(disposable);
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onNext(Response<R> response) {
|
|
if (response.isSuccessful()) {
|
|
this.observer.onNext(response.body());
|
|
return;
|
|
}
|
|
this.terminated = true;
|
|
HttpException httpException = new HttpException(response);
|
|
try {
|
|
this.observer.onError(httpException);
|
|
} catch (Throwable th) {
|
|
Exceptions.b(th);
|
|
RxJavaPlugins.b(new CompositeException(httpException, th));
|
|
}
|
|
}
|
|
}
|
|
|
|
BodyObservable(Observable<Response<T>> observable) {
|
|
this.upstream = observable;
|
|
}
|
|
|
|
@Override // io.reactivex.Observable
|
|
protected void subscribeActual(Observer<? super T> observer) {
|
|
this.upstream.subscribe(new BodyObserver(observer));
|
|
}
|
|
}
|