RxJava之from系列操作符源码解析

转载请以链接形式标明出处：
本文出自:103style的博客

from 系列操作符

包括以下操作符：

  • fromArray(T... items) : 参数 数组长度 为 0 是执行 empty操作符，长度为 1 时，是执行 just操作符。 
    public static <T> Observable<T> fromArray(T... items) {
    if (items.length == 0) {
        return empty();
    } else
    if (items.length == 1) {
        return just(items[0]);
    }
    return RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items));
}
  • fromIterable(Iterable<? extends T> source)
  • fromCallable(Callable<? extends T> supplier)
  • fromFuture(Future<? extends T> future)
  • fromPublisher(Publisher<? extends T> publisher)

from 操作符实际上返回的是一个 ObservableFromXXX对象。（ XXX 代表 Array,Iterable,Callable,Future,Publisher）

ObservableFromArray源码：

  • 首先构建 FromArrayDisposable 对象
  • 然后调用观察者的 onSubscribe(Disposable d) 方法。
  • FromArrayDisposable 的 fusionMode 默认为 false， 所以继续执行 FromArrayDisposable 的 run 方法。
  • FromArrayDisposable 的 run 方法：依次传入参数数组中的值到 观察者 的 onNext 方法， 如果某个值为 null， 直接 onError 结束，否则遍历完之后，执行 onComplete()；
public final class ObservableFromArray<T> extends Observable<T> {
    final T[] array;
    public ObservableFromArray(T[] array) {
        this.array = array;
    }

    @Override
    public void subscribeActual(Observer<? super T> observer) {
        FromArrayDisposable<T> d = new FromArrayDisposable<T>(observer, array);
        observer.onSubscribe(d);
        if (d.fusionMode) {
            return;
        }
        d.run();
    }

    static final class FromArrayDisposable<T> extends BasicQueueDisposable<T> {
        final Observer<? super T> downstream;
        final T[] array;
        boolean fusionMode;
            ...
        FromArrayDisposable(Observer<? super T> actual, T[] array) {
            this.downstream = actual;
            this.array = array;
        }
            ...
        void run() {
            T[] a = array;
            int n = a.length;

            for (int i = 0; i < n && !isDisposed(); i++) {
                T value = a[i];
                if (value == null) {
                    downstream.onError(new NullPointerException("The element at index " + i + " is null"));
                    return;
                }
                downstream.onNext(value);
            }
            if (!isDisposed()) {
                downstream.onComplete();
            }
        }
    }
}

ObservableFromIterable 和 ObservableFromArray类似：

  • 长度为0 直接调用 观察者 的 onSubscribe(Disposable d) 和 onComplete() 方法结束。
  • 然后依次传入Iterator 中的值到 观察者 的 onNext 方法， 如果某个值为 null， 直接 onError 结束，否则遍历完之后，执行 onComplete()；
public void subscribeActual(Observer<? super T> observer) {
    Iterator<? extends T> it;
    try {
        it = source.iterator();
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        EmptyDisposable.error(e, observer);
        return;
    }
    boolean hasNext;
    try {
        hasNext = it.hasNext();
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        EmptyDisposable.error(e, observer);
        return;
    }
    if (!hasNext) {
        EmptyDisposable.complete(observer);
        return;
    }

    FromIterableDisposable<T> d = new FromIterableDisposable<T>(observer, it);
    observer.onSubscribe(d);

    if (!d.fusionMode) {
        d.run();
    }
}

void run() {
    boolean hasNext;
    do {
        ...
        T v;
        try {
            v = ObjectHelper.requireNonNull(it.next(), "The iterator returned a null value");
        } catch (Throwable e) {
            downstream.onError(e);
            return;
        }
        downstream.onNext(v);
        ...
        try {
            hasNext = it.hasNext();
        } catch (Throwable e) {
            downstream.onError(e);
            return;
        }
    } while (hasNext);

    if (!isDisposed()) {
        downstream.onComplete();
    }
}

ObservableFromCallable 相关源码：

  • 首先构建 DeferredScalarDisposable 对象。
  • 调用 观察者 的 onSubscribe(Disposable d) 方法。
  • 获取 Callable.call() 的返回值，传给 DeferredScalarDisposable 对象的 complete(T value) 方法。
    @Override
    public void subscribeActual(Observer<? super T> observer) {
        DeferredScalarDisposable<T> d = new DeferredScalarDisposable<T>(observer);
        observer.onSubscribe(d);
        if (d.isDisposed()) {
            return;
        }
        T value;
        try {
            value = ObjectHelper.requireNonNull(callable.call(), "Callable returned null");
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            if (!d.isDisposed()) {
                observer.onError(e);
            } else {
                RxJavaPlugins.onError(e);
            }
            return;
        }
        d.complete(value);
    }

DeferredScalarDisposable 的 complete(T value) 方法

  • get() 默认值为 0, 所以 state == FUSED_EMPTY 条件不成立，所以设置为 TERMINATED状态， 然后调用 观察者 的 onNext(T value) 方法。
  • 然后再调用 观察者 的 ononComplete() 方法。
static final int TERMINATED = 2;
static final int DISPOSED = 4;
static final int FUSED_EMPTY = 8;
static final int FUSED_READY = 16;
static final int FUSED_CONSUMED = 32;

public final void complete(T value) {
    int state = get();
    if ((state & (FUSED_READY | FUSED_CONSUMED | TERMINATED | DISPOSED)) != 0) {
        return;
    }
    Observer<? super T> a = downstream;
    if (state == FUSED_EMPTY) {
        this.value = value;
        lazySet(FUSED_READY);
        a.onNext(null);
    } else {
        lazySet(TERMINATED);
        a.onNext(value);
    }
    if (get() != DISPOSED) {
        a.onComplete();
    }
}

ObservableFromFuture源码：

  • 首先构建 DeferredScalarDisposable 对象。
  • 调用 观察者 的 onSubscribe(Disposable d) 方法。
  • 获取 future.get(timeout, unit) 这个 future.get() 的返回值，
    传给 DeferredScalarDisposable 对象的 complete(T value) 方法。(和ObservableFromCallable 相同)
public void subscribeActual(Observer<? super T> observer) {
    DeferredScalarDisposable<T> d = new DeferredScalarDisposable<T>(observer);
    observer.onSubscribe(d);
    if (!d.isDisposed()) {
        T v;
        try {
            v = ObjectHelper.requireNonNull(unit != null ? future.get(timeout, unit) : future.get(), "Future returned null");
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            if (!d.isDisposed()) {
                observer.onError(ex);
            }
            return;
        }
        d.complete(v);
    }
}

RxJava提供的 FutureObserver类 的 get 方法：

public final class FutureObserver<T> extends CountDownLatch
        implements Observer<T>, Future<T>, Disposable {
        ...
    public T get() throws InterruptedException, ExecutionException {
        if (getCount() != 0) {
            BlockingHelper.verifyNonBlocking();
            await();
        }

        if (isCancelled()) {
            throw new CancellationException();
        }
        Throwable ex = error;
        if (ex != null) {
            throw new ExecutionException(ex);
        }
        return value;
    }

    @Override
    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        if (getCount() != 0) {
            BlockingHelper.verifyNonBlocking();
            if (!await(timeout, unit)) {
                throw new TimeoutException(timeoutMessage(timeout, unit));
            }
        }

        if (isCancelled()) {
            throw new CancellationException();
        }

        Throwable ex = error;
        if (ex != null) {
            throw new ExecutionException(ex);
        }
        return value;
    }  
        ...
}

ObservableFromPublisher源码：

  • 构建了一个 PublisherSubscriber 对象，重写了 观察者 的 onSubscribe 方法。
public final class ObservableFromPublisher<T> extends Observable<T> {

    final Publisher<? extends T> source;

    public ObservableFromPublisher(Publisher<? extends T> publisher) {
        this.source = publisher;
    }

    @Override
    protected void subscribeActual(final Observer<? super T> o) {
        source.subscribe(new PublisherSubscriber<T>(o));
    }

    static final class PublisherSubscriber<T>
            implements FlowableSubscriber<T>, Disposable {

        final Observer<? super T> downstream;
        Subscription upstream;

        PublisherSubscriber(Observer<? super T> o) {
            this.downstream = o;
        }
            ...
        @Override
        public void onSubscribe(Subscription s) {
            if (SubscriptionHelper.validate(this.upstream, s)) {
                this.upstream = s;
                downstream.onSubscribe(this);
                s.request(Long.MAX_VALUE);
            }
        }
            ...
    }
}

SubscriptionHelper 的 validate(Subscription current, Subscription next) 方法：

  • onSubscribe 传参不能为 null。
  • upstream默认为 null，所以第一次调用直接返回 true。
  • 当第二次或者多次 调用 onSubscribe 方法时，if (current != null) 条件成立，直接调用 第二次传参的 cancel 方法。
    然后直接结束当前订阅流程。
public enum SubscriptionHelper implements Subscription {
        ...
    public static boolean validate(Subscription current, Subscription next) {
        if (next == null) {
            RxJavaPlugins.onError(new NullPointerException("next is null"));
            return false;
        }
        if (current != null) {
            next.cancel();
            reportSubscriptionSet();
            return false;
        }
        return true;
    }
        ...
}

以上

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 215,384评论 6赞 497
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 91,845评论 3赞 391
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 161,148评论 0赞 351
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 57,640评论 1赞 290
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 66,731评论 6赞 388
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,712评论 1赞 294
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,703评论 3赞 415
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 38,473评论 0赞 270
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,915评论 1赞 307
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,227评论 2赞 331
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,384评论 1赞 345
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 35,063评论 5赞 340
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,706评论 3赞 324
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,302评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,531评论 1赞 268
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 47,321评论 2赞 368
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 44,248评论 2赞 352