Java 类io.reactivex.internal.functions.ObjectHelper 实例源码

@Override
protected void subscribeActual(CompletableObserver s) {
    CompletableObserver observer;
    try {
        observer = ObjectHelper.requireNonNull(s, "Null Observer");
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Disposable already
        RxJavaPlugins.onError(e);

        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }

    source.subscribe(new AttachCompletableObserver(observer, this.compositeDisposable));
}

@Override
protected void subscribeActual(Subscriber<? super T> s) {

    Subscriber<? super T> subscriber;
    try {
        subscriber = ObjectHelper.requireNonNull(s, "Null Observer");
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Disposable already
        RxJavaPlugins.onError(e);

        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }

    source.subscribe(new AttachSingleObserver<>(subscriber, this.compositeDisposable));
}

@Override
public void subscribeActual(Observer<? super T> s) {
    Observer<? super T> observer;
    try {
        observer = ObjectHelper.requireNonNull(s, "Null Observer");
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Disposable already
        RxJavaPlugins.onError(e);

        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }

    source.subscribe(new AttachSingleObserver<>(observer, this.compositeDisposable));
}

@Override
protected void subscribeActual(MaybeObserver<? super T> s) {
    MaybeObserver<? super T> observer;
    try {
        observer = ObjectHelper.requireNonNull(s, "Null Observer");
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Disposable already
        RxJavaPlugins.onError(e);

        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }

    source.subscribe(new AttachMaybeObserver<>(observer, this.compositeDisposable));
}

@Override
protected void subscribeActual(SingleObserver<? super T> s) {
    SingleObserver<? super T> observer;
    try {
        observer = ObjectHelper.requireNonNull(s, "Null Observer");
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Disposable already
        RxJavaPlugins.onError(e);

        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }

    source.subscribe(new AttachSingleObserver<>(observer, this.compositeDisposable));
}

@Override
public void onNext(T t) {
    if (done) {
        return;
    }

    if (sourceMode == ASYNC) {
        actual.onNext(null);
        return;
    }

    Optional<R> o;

    try {
        o = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null Optional");
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        fail(ex);
        return;
    }

    if (o.isPresent()) {
        actual.onNext(o.get());
    }
}

@Override
public boolean tryOnNext(T t) {
    if (done) {
        return false;
    }

    if (sourceMode == ASYNC) {
        return actual.tryOnNext(null);
    }

    Optional<R> o;

    try {
        o = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null Optional");
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        fail(ex);
        return false;
    }

    if (o.isPresent()) {
        return actual.tryOnNext(o.get());
    }
    return false;
}

@Override
public void onSuccess(T value) {
    Optional<R> v;

    try {
        v = ObjectHelper.requireNonNull(mapper.apply(value), "The mapper returned a null Optional");
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        actual.onError(ex);
        return;
    }

    if (v.isPresent()) {
        actual.onSuccess(v.get());
    } else {
        actual.onComplete();
    }
}

/**
 * The given consumer is invoked with each registered handler instance.
 * <p>
 * Exceptions raised by the invocation of the consumer for a particular
 * handler are printed to the console and the current thread's
 * uncaught exception handler is notified.
 * <p>
 * This method is threadsafe.
 * @param consumer the consumer to invoke
 */
public final void forEach(@NonNull Consumer<H> consumer) {
    ObjectHelper.requireNonNull(consumer, "consumer is null");
    Iterator<HandlerRegistration<H>> it = handlers.iterator();
    while (it.hasNext()) {
        try {
            HandlerRegistration<H> hr = it.next();
            H h = hr.get();
            if (h != null) {
                consumer.accept(h);
            }
        } catch (Throwable ex) {
            ex.printStackTrace();
            Thread t = Thread.currentThread();
            t.getUncaughtExceptionHandler().uncaughtException(t, ex);
        }
    }
}

/**
 * The given consumer is invoked with each registered handler instance.
 * <p>
 * Exceptions raised by the invocation of the consumer for a particular
 * handler are printed to the console and the current thread's
 * uncaught exception handler is notified.
 * <p>
 * This method is threadsafe.
 * @param <S> the type of the extra state provided to the consumer
 * @param state the extra state provided to the consumer
 * @param consumer the consumer to invoke
 */
public final <S> void forEach(S state, @NonNull BiConsumer<S, H> consumer) {
    ObjectHelper.requireNonNull(consumer, "consumer is null");
    Iterator<HandlerRegistration<H>> it = handlers.iterator();
    while (it.hasNext()) {
        try {
            HandlerRegistration<H> hr = it.next();
            H h = hr.get();
            if (h != null) {
                consumer.accept(state, h);
            }
        } catch (Throwable ex) {
            ex.printStackTrace();
            Thread t = Thread.currentThread();
            t.getUncaughtExceptionHandler().uncaughtException(t, ex);
        }
    }
}

@Override
public void onNext(T t) {
    Publisher<? extends R> p;
    try {
        p = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null Publisher");
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        s.cancel();
        onError(ex);
        return;
    }

    SfmInnerSubscriber<T, R> inner = new SfmInnerSubscriber<T, R>(this, bufferSize);
    if (add(inner)) {
        p.subscribe(inner);
    }
}

void fastpath() {
    Callable<T> c = callable;
    for (;;) {
        if (cancelled) {
            break;
        }

        T v;

        try {
            v = ObjectHelper.requireNonNull(c.call(), "The callable returned a null value");
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            actual.onError(ex);
            break;
        }

        actual.onNext(v);
    }
}

void fastpath() {
    Callable<T> c = callable;
    for (;;) {
        if (cancelled) {
            break;
        }

        T v;

        try {
            v = ObjectHelper.requireNonNull(c.call(), "The callable returned a null value");
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            actual.onError(ex);
            break;
        }

        actual.tryOnNext(v);
    }
}

@Override
public void onNext(T t) {
    produced++;
    actual.onNext(t);

    Publisher<? extends T> p;
    try {
        p = ObjectHelper.requireNonNull(expander.apply(t), "The expander returned a null Publisher");
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        super.cancel();
        actual.onError(ex);
        drainQueue();
        return;
    }

    queue.offer(p);
}

@Override
protected void subscribeActual(Subscriber<? super Void> s) {
    Iterator<? extends Nono> it;

    try {
        it = ObjectHelper.requireNonNull(sources.iterator(), "The sources Iterable returned a null Iterator");
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        EmptySubscription.error(ex, s);
        return;
    }

    ConcatSubscriber parent = new ConcatSubscriber(s, it, delayError);
    s.onSubscribe(parent);
    parent.drain();
}

@Override
public void onNext(T t) {
    if (!done) {
        if (sourceMode == NONE) {
            R v;

            try {
                v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null value");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }

            actual.onNext(v);
        } else {
            actual.onNext(null);
        }
    }
}

@Override
protected void subscribeActual(Subscriber<? super T> s) {
    FlowableProcessor<Throwable> pp = PublishProcessor.<Throwable>create().toSerialized();

    Publisher<?> when;
    try {
        when = ObjectHelper.requireNonNull(handler.apply(pp), "The handler returned a null Publisher");
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        EmptySubscription.error(ex, s);
        return;
    }

    RetrySubscriber<T> parent = new RetrySubscriber<T>(s, pp, source);
    s.onSubscribe(parent);

    when.subscribe(parent.other);
    parent.subscribeNext();
}

@Override
protected void subscribeActual(Subscriber<? super T> s) {
    FlowableProcessor<Throwable> pp = PublishProcessor.<Throwable>create().toSerialized();

    Publisher<?> when;
    try {
        when = ObjectHelper.requireNonNull(handler.apply(pp), "The handler returned a null Publisher");
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        EmptySubscription.error(ex, s);
        return;
    }

    RetrySubscriber<T> parent = new RetrySubscriber<T>(s, pp, source);
    s.onSubscribe(parent);

    when.subscribe(parent.other);
    parent.subscribeNext();
}

void onSuccess(int index, T value) {
    values[index] = value;
    if (wip.decrementAndGet() == 0) {
        R v;
        try {
            v = ObjectHelper.requireNonNull(zipper.apply(values), "The zipper returned a null value");
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            Arrays.fill(values, null);
            actual.onError(ex);
            return;
        }

        Arrays.fill(values, null);
        complete(v);
    }
}

@Override
public void onNext(In t) {
    if (done) {
        return;
    }
    if (!createdState()) {
        return;
    }
    if (--count == 0) {
        requestsArrived = true;
        count = requestBatchSize;
    }
    try {
        drainCalled = false;
        state = ObjectHelper.requireNonNull(transition.apply(state, t, this),
                "intermediate state cannot be null");
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        onError(e);
        return;
    }
    if (!drainCalled) {
        drain();
    }
}

private boolean createdState() {
    if (state == null) {
        try {
            state = ObjectHelper.requireNonNull(initialState.call(),
                    "initial state cannot be null");
            return true;
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            done = true;
            onError_(e);
            return false;
        }
    } else {
        return true;
    }
}

@Override
public void onNext(T t) {
    if (done) {
        ResourceFlowable.releaseItem(t, release);
    } else {
        R v;

        try {
            v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null value");
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            upstream.cancel();
            ResourceFlowable.releaseItem(t, release);
            done = true;
            actual.onError(ex);
            return;
        }

        ResourceFlowable.releaseItem(t, release);

        actual.onNext(v);
    }
}

@Override
public void onNext(T t) {
    if (done) {
        ResourceFlowable.releaseItem(t, release);
    } else {
        R v;

        try {
            v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null value");
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            upstream.cancel();
            ResourceFlowable.releaseItem(t, release);
            done = true;
            actual.onError(ex);
            return;
        }

        ResourceFlowable.releaseItem(t, release);

        actual.onNext(v);
    }
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<ActionEvent> actions(@NonNull AbstractButton button) {
    ObjectHelper.requireNonNull(button, "button is null");
    return RxSwingPlugins.onAssembly(new ActionEventObservable(button));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<ActionEvent> actions(@NonNull JComboBox<?> button) {
    ObjectHelper.requireNonNull(button, "button is null");
    return RxSwingPlugins.onAssembly(new ActionEventComboBoxObservable(button));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<MouseEvent> mouse(@NonNull Component component, int flags) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new MouseEventObservable(component, flags));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<MouseWheelEvent> mouseWheel(@NonNull Component component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new MouseWheelEventObservable(component));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<KeyEvent> keyboard(@NonNull Component component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new KeyEventObservable(component));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<FocusEvent> focus(@NonNull Component component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new FocusEventObservable(component));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<HierarchyEvent> hierarchyBounds(@NonNull Component component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new HierarchyBoundsEventObservable(component));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<HierarchyEvent> hierarchy(@NonNull Component component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new HierarchyEventObservable(component));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<InputMethodEvent> inputMethod(@NonNull Component component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new InputMethodEventObservable(component));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<PropertyChangeEvent> propertyChange(@NonNull Component component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new PropertyChangeEventObservable(component, null));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<PropertyChangeEvent> propertyChange(@NonNull Component component, String propertyName) {
    ObjectHelper.requireNonNull(component, "component is null");
    ObjectHelper.requireNonNull(propertyName, "propertyName is null");
    return RxSwingPlugins.onAssembly(new PropertyChangeEventObservable(component, propertyName));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<DocumentEvent> document(@NonNull Document component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new DocumentEventObservable(component));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<DocumentEvent> document(@NonNull JTextComponent component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return document(component.getDocument());
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<UndoableEditEvent> undoableEdit(@NonNull Document component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new UndoableEditEventObservable(component));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<CaretEvent> caret(@NonNull JTextComponent component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new CaretEventObservable(component));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<AncestorEvent> ancestor(@NonNull JComponent component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new AncestorEventObservable(component));
}

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable<VetoablePropertyChangeEvent> vetoableChange(@NonNull JComponent component) {
    ObjectHelper.requireNonNull(component, "component is null");
    return RxSwingPlugins.onAssembly(new VetoableChangeEventObservable(component));
}