Jiacheng Zheng

[zhenglei21887]

No description provided.

[leonchen83]

    public Observable<String> concat(List<Observable<String>> source) {
        return Observable.create(emitter -> { concat(source, emitter); });
    }

    private void concat(List<Observable<String>> source, ObservableEmitter<String> emitter) {
        if (source.isEmpty()) {
            emitter.onComplete();
        } else {
            source.get(0).subscribe(e -> {
                emitter.onNext(e);
            }, e -> emitter.onError(e), () -> {
                concat(source.subList(1, source.size()), emitter);
            });
        }
    }

    public Observable<String> delayAll(Observable<String> source, long time, TimeUnit unit) {
        return source.concatMap(e -> Observable.just(e).delay(time, unit));
    }

    public Observable<String> distinct(Observable<String> source) {
        return source.groupBy(e -> e).map(e -> e.getKey());
    }

    public Observable<String> repeat(Observable<String> source, int count) {
        return Observable.range(0, count).concatMap(e -> source);
    }

    public Observable<Integer> iterate(Observable<Node> observable) {
        return observable.flatMap(n -> {
            Observable<Integer> left = n.left == null ? Observable.empty() : iterate(Observable.just(n.left));
            Observable<Integer> right = n.right == null ? Observable.empty() : iterate(Observable.just(n.right));
            return Observable.merge(left, right, Observable.just(n.value));
        });
    }