io.reactivex.rxjava3.core.MaybeObserver#onError ( )源码实例Demo

@Override
protected void subscribeActual(MaybeObserver<? super T> downstream) {
    if (circuitBreaker.tryAcquirePermission()) {
        upstream.subscribe(new CircuitBreakerMaybeObserver(downstream));
    } else {
        downstream.onSubscribe(EmptyDisposable.INSTANCE);
        downstream.onError(createCallNotPermittedException(circuitBreaker));
    }
}
 

@Override
protected void subscribeActual(MaybeObserver<? super T> downstream) {
    if (bulkhead.tryAcquirePermission()) {
        upstream.subscribe(new BulkheadMaybeObserver(downstream));
    } else {
        downstream.onSubscribe(EmptyDisposable.INSTANCE);
        downstream.onError(BulkheadFullException.createBulkheadFullException(bulkhead));
    }
}
 

@Override
protected void subscribeActual(MaybeObserver<? super T> downstream) {
    long waitDuration = rateLimiter.reservePermission();
    if (waitDuration >= 0) {
        if (waitDuration > 0) {
            Completable.timer(waitDuration, TimeUnit.NANOSECONDS)
                .subscribe(() -> upstream.subscribe(new RateLimiterMaybeObserver(downstream)));
        } else {
            upstream.subscribe(new RateLimiterMaybeObserver(downstream));
        }
    } else {
        downstream.onSubscribe(EmptyDisposable.INSTANCE);
        downstream.onError(RequestNotPermitted.createRequestNotPermitted(rateLimiter));
    }
}