RxJava之from操作符

1
2
3
4
5
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
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
    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);
        }
    }