Rxjava Any

所有代码的演示都在RxJava2.2.4版本上进行的

判断数据是否有满足条件的

1.example

          Observable.just(11, 12, 13, 14, 15)
                .any(Functions.equalsWith(15))
                .subscribe(result -> Logger.getGlobal().info("result:" + result));

2.源码分析

 public final Single<Boolean> any(Predicate<? super T> predicate) {
        ObjectHelper.requireNonNull(predicate, "predicate is null");
        return RxJavaPlugins.onAssembly(new ObservableAnySingle<T>(this, predicate));
    }

调用any会产生Single，只发射单个数据或错误事件。
Predicate的实现类,Predicate接口比较简单，测试一下给定数据

Predicate.png

public interface Predicate<T> {
    /**
     * Test the given input value and return a boolean.
     * @param t the value
     * @return the boolean result
     * @throws Exception on error
     */
    boolean test(@NonNull T t) throws Exception;
}

看看subscribe方法

    public final Disposable subscribe(final Consumer<? super T> onSuccess, final Consumer<? super Throwable> onError) {
        ObjectHelper.requireNonNull(onSuccess, "onSuccess is null");
        ObjectHelper.requireNonNull(onError, "onError is null");

        ConsumerSingleObserver<T> observer = new ConsumerSingleObserver<T>(onSuccess, onError);
        subscribe(observer);
        return observer;
    }

构造一个ConsumerSingleObserver返回，
ConsumerSingleObserver中有2个属性，看的出Single只关心要么成功，要么失败

 final Consumer<? super T> onSuccess;

 final Consumer<? super Throwable> onError;

在这个demo中调用subscribe实际上触发ObservableAnySingle实例的subscribeActual方法

 protected void subscribeActual(SingleObserver<? super Boolean> t) {
        source.subscribe(new AnyObserver<T>(t, predicate));
    }

这里的source就是ConsumerSingleObserver的实例
重点来分析

   static final class AnyObserver<T> implements Observer<T>, Disposable {
        //downstream就是上面的ConsumerSingleObserver实例
        final SingleObserver<? super Boolean> downstream;
        //predicate就是Functions.equalsWith(15)的实例
        final Predicate<? super T> predicate;

        Disposable upstream;

        boolean done;

        AnyObserver(SingleObserver<? super Boolean> actual, Predicate<? super T> predicate) {
            this.downstream = actual;
            this.predicate = predicate;
        }

        @Override
        public void onSubscribe(Disposable d) {
            if (DisposableHelper.validate(this.upstream, d)) {
                this.upstream = d;
                downstream.onSubscribe(this);
            }
        }

        @Override
        public void onNext(T t) {
            if (done) {
              //已经断言成功，直接返回
                return;
            }
            boolean b;
            try {
              //重点在这
                b = predicate.test(t);
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                upstream.dispose();
                onError(e);
                return;
            }
            if (b) {
                done = true;
                upstream.dispose();
                downstream.onSuccess(true);
            }
        }

        @Override
        public void onError(Throwable t) {
            if (done) {
                RxJavaPlugins.onError(t);
                return;
            }

            done = true;
            downstream.onError(t);
        }

        @Override
        public void onComplete() {
            if (!done) {
                done = true;
                downstream.onSuccess(false);
            }
        }

        @Override
        public void dispose() {
            upstream.dispose();
        }

        @Override
        public boolean isDisposed() {
            return upstream.isDisposed();
        }
    }

onNext方法每次都给被调用，首次先检查是否已经完成，因为比如有[1,2,3,4,5,6,7,8,9]数据，
你想检查是否含有10，显然没有，数据会经过1，2，3，4...一直到9，最后都没有10，此时的
done == false，数据已经结束，调用onComplete，返回false；如果你想检查2，数据到1，发现
测试失败，然后到2，测试成功，done == true，上流upstream就会停止