RxJava之from操作符

57 阅读 0 评论 38 点赞

我是靠谱客的博主细心项链，最近开发中收集的这篇文章主要介绍RxJava之from操作符，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

关于callable与Future写在最前面详见这儿

from操作符

一、作用

from操作符与just操作符类似，都用来把参数转为Observable对象。其官方示意图为：

由上图所示，from操作符将参数转为事件流来传递（下方横向箭头表示事件流）

二、用法

1. Observable.from(T[] array);
2. Observable.from(@NotNull Iterable<?> iterable);
3. Observable.from(Future<?> future)；
4. Observable.from(Future<?> future, long timeout/*超时时间*/, TimeUnit unit);//
5. Observable.from(Future<?> future, Scheduler scheduler);

三、事件流分析

其中用法1和2，与just操作符的事件流分发机制一样。[参见此处]
用法3，4，5

事件流如下图所示

这里写图片描述

    public final static <T> Observable<T> from(Future<? extends T> future) {
    //创建OnSubscribe对象
    return create(OnSubscribeToObservableFuture.toObservableFuture(future));
    }

    //下面是该OnSubscribe
    OnSubscribeToObservableFuture.ToObservableFuture// implements OnSubscribe
    //该OnSubscribe的call方法如下：

    @Override
    public void call(Subscriber<? super T> subscriber) {
        subscriber.add(Subscriptions.create(new Action0() {
            @Override
            public void call() {
                // If the Future is already completed, "cancel" does nothing.
                that.cancel(true);
            }
        }));
        try {
                //don't block or propagate CancellationException if already unsubscribed
                if (subscriber.isUnsubscribed()) {
                    return;
                }
                T value = (unit == null) ? (T) that.get() : (T) that.get(time, unit);
                subscriber.onNext(value);
                subscriber.onCompleted();
            } catch (Throwable e) {
                // If this Observable is unsubscribed, we will receive an CancellationException.
             // However, CancellationException will not be passed to the final Subscriber
             // since it's already subscribed.
             // If the Future is canceled in other place, CancellationException will be still
             // passed to the final Subscriber.
             if (subscriber.isUnsubscribed()) {
                //refuse to emit onError if already unsubscribed
                return;
            }
            Exceptions.throwOrReport(e, subscriber);
        }
    }

总结

Observalbe.from的简单用法及事件流与just操作符一致，其Future用法总结如下：
1. 当observer订阅到Observable时，此时Observable已经保留了OnSubscribeToObservableFuture.ToObservable的实例，并在subscribe（Observer， Observable）方法中回调onSubscribe的call方法（）；
2. 在OnSubscribe的call方法中，取得future的值，并发送到Observer中。