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中。