RxJava 1.x 笔记：创建型操作符

//使用 create 方法创造被观察者
mObservable = Observable.create(new Observable.OnSubscribe<String>() {
    /**
     * 当 observable 被订阅时，会自动调用 call() 方法，依次触发其中的事件
     * 其实就是调用订阅者的回调方法，即实现了被观察者向观察者的事件传递
     * @param subscriber
     */
    @Override
    public void call(Subscriber<? super String> subscriber) {
        if (!subscriber.isUnsubscribed()) {
            //这里可能会有复杂的事件发送规则
            for (int i = 0; i < 5; i ++) {
                subscriber.onNext("next 1");
            }
            subscriber.onCompleted();   //必须得调用 completed
        }
        //意外情况下，调用 onError
        subscriber.onError(new Throwable("Error message"));
    }
});

public final static <T> Observable<T> defer(Func0<Observable<T>> observableFactory) {
    return create(new OnSubscribeDefer<T>(observableFactory));
}

mObservable = Observable.defer(new Func0<Observable<String>>() {
    @Override
    public Observable<String> call() {
        return "test";
    }
});

public final static <T> Observable<T> empty() {
    return (Observable<T>) EmptyHolder.INSTANCE;
}
//懒加载、静态内部类单例，只发射一个 onCompleted()
private static final class EmptyHolder {
    final static Observable<Object> INSTANCE = create(new OnSubscribe<Object>() {
        @Override
        public void call(Subscriber<? super Object> subscriber) {
            subscriber.onCompleted();
        }
    });
}

//创建一个不发射任何数据，只会发射 onCompleted()
mObservable = Observable.empty();

public final static <T> Observable<T> never() {
    return NeverObservable.instance();
}
private static class NeverObservable<T> extends Observable<T> {

    private static class Holder {
        static final NeverObservable<?> INSTANCE = new NeverObservable<Object>();
    }

    @SuppressWarnings("unchecked")
    static <T> NeverObservable<T> instance() {
        return (NeverObservable<T>) Holder.INSTANCE;
    }

    NeverObservable() {
        super(new OnSubscribe<T>() {
            @Override
            public void call(Subscriber<? super T> observer) {
                // do nothing
            }

        });
    }
}

//不会发出任何消息
mObservable = Observable.never();

public final static <T> Observable<T> error(Throwable exception) {
    return new ThrowObservable<T>(exception);
}
private static class ThrowObservable<T> extends Observable<T> {

    public ThrowObservable(final Throwable exception) {
        super(new OnSubscribe<T>() {

            @Override
            public void call(Subscriber<? super T> observer) {
                observer.onError(exception);
            }

        });
    }
}

//只会发出错误消息
mObservable = Observable.error(new Throwable("error message"));

public final static <T> Observable<T> from(T[] array) {
    return from(Arrays.asList(array));
}
public final static <T> Observable<T> from(Iterable<? extends T> iterable) {
    return create(new OnSubscribeFromIterable<T>(iterable));
}
public static <T> Observable<T> from(Future<? extends T> future) {
    return (Observable<T>)unsafeCreate(OnSubscribeToObservableFuture.toObservableFuture(future));
}

private void createObservables() {
    String[] words = {"shixin", "is", "cute"};
    Observable<String> from = Observable.from(words);
    from.subscribe(new Subscriber<String>() {
        @Override
        public void onCompleted() {
            System.out.println("onCompleted");
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onNext(String s) {
            System.out.println("onNext: " + s);
        }
    });
}

public static Observable<Long> interval(long interval, TimeUnit unit) {
    return interval(interval, interval, unit, Schedulers.computation());
}
public static Observable<Long> interval(long initialDelay, long period, TimeUnit unit, Scheduler scheduler) {
    return unsafeCreate(new OnSubscribeTimerPeriodically(initialDelay, period, unit, scheduler));
}

private void createObservableWithInterval() {
    Observable<Long> interval = Observable.interval(1, TimeUnit.SECONDS);
    interval.subscribe(getPrintSubscriber());
}

/**
 * 用于打印结果的订阅者
 * @param <T>
 * @return
 */
private <T> Subscriber<T> getPrintSubscriber() {
    return new Subscriber<T>() {
        @Override
        public void onCompleted() {
            System.out.println("onCompleted");
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("onError: " + e.getMessage());
        }

        @Override
        public void onNext(T t) {
            System.out.println("onNext: " + t);
            if ((Long)t > 10){
                unsubscribe();
            }
        }
    };
}

public static <T> Observable<T> just(final T value) {
    return ScalarSynchronousObservable.create(value);
}
public static <T> Observable<T> just(T t1, T t2) {
    return from((T[])new Object[] { t1, t2 });
}

private void createObservableWithJust() {
    String[] words = {"shixin", "is", "cute"};
    Observable<String[]> just = Observable.just(words);
    just.subscribe(getPrintSubscriber());
}

public static Observable<Integer> range(int start, int count) {
    if (count < 0) {
        throw new IllegalArgumentException("Count can not be negative");
    }
    if (count == 0) {
        return Observable.empty();
    }
    if (start > Integer.MAX_VALUE - count + 1) {
        throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");
    }
    if (count == 1) {
        return Observable.just(start);
    }
    return Observable.unsafeCreate(new OnSubscribeRange(start, start + (count - 1)));
}

private void createObservableWithRange() {
    Observable<Integer> range = Observable.range(3, 5);
    range.subscribe(this.<Integer>getPrintSubscriber());
}

public final Observable<T> repeat(final long count) {
    return OnSubscribeRedo.<T>repeat(this, count);
}

private void createObservableWithRepeat() {
    String[] words = {"shixin", "is", "cute"};
    Observable<String> from = Observable.from(words);
    from.repeat(2)
            .subscribe(getPrintSubscriber());
}

public final Observable<T> repeatWhen(final Func1<? super Observable<? extends Void>, ? extends Observable<?>> notificationHandler) {
    return OnSubscribeRedo.repeat(this, InternalObservableUtils.createRepeatDematerializer(notificationHandler));
}

Observable.unsafeCreate(new Observable.OnSubscribe<String>() {
    @Override
    public void call(final Subscriber<? super String> subscriber) {
        for (int i = 0; i < 5; i++) {
                subscriber.onNext("item " + i);
        }
        subscriber.onCompleted();
    }
}).repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
    @Override
    public Observable<?> call(final Observable<? extends Void> completed) {
        //每次调用 onCompleted，都会进入这里，需要在这里决定是否需要重订阅
        return completed.delay(5, TimeUnit.SECONDS);
    }
}).subscribe(getPrintSubscriber());

public static Observable<Long> timer(long delay, TimeUnit unit) {
    return timer(delay, unit, Schedulers.computation());
}
public static Observable<Long> timer(long delay, TimeUnit unit, Scheduler scheduler) {
    return unsafeCreate(new OnSubscribeTimerOnce(delay, unit, scheduler));
}

private void createObservableWithTimer() {
    Observable.timer(3, TimeUnit.SECONDS)
            .subscribe(this.<Long>getPrintSubscriber());
}