两个有序可观察量的完全外连接

假设我们有两个可观察量Observable<Integer> o1 and Observable<Integer> o2每个可观察量都产生严格递增的序列。

任务是对这两个可观察量执行相当于完全外连接的操作。例如加入

    Observable.just(0, 2, 3, 6)
    Observable.just(1, 2, 3, 4, 5, 6)

应该产生

    [ [0, _], [_, 1], [2, 2], [3, 3], [_, 4], [_, 5], [6, 6] ]

连接应该高效并且在非常大或无限的流上运行良好。

该解决方案在拉动场景中很容易。有没有一种惯用的 rx 方法来实现这一目标？

没有单一的运算符可以实现此目的，但可以从标准和扩展运算符组成行为：

static abstract class Pair implements Comparable<Pair> { 
    int value;

    @Override
    public int compareTo(Pair o) {
        return Integer.compare(value, o.value);
    }
}

static final class Left extends Pair {
    Left(int value) {
        this.value = value;
    }

    @Override
    public String toString() {
        return "[" + value + ", _]";
    }
}

static final class Right extends Pair {
    Right(int value) {
        this.value = value;
    }

    @Override
    public String toString() {
        return "[_, " + value + "]";
    }
}

static final class Both extends Pair {
    Both(int value) {
        this.value = value;
    }

    @Override
    public int hashCode() {
        return value;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof Both) {
            return ((Both)obj).value == value;
        }
        return false;
    }

    @Override
    public String toString() {
        return "[" + value + ", " + value + "]";
    }
}

@SuppressWarnings("unchecked")
@Test
public void test() {
    Flowable<Integer> a = Flowable.just(0, 2, 3, 6);
    Flowable<Integer> b = Flowable.just(1, 2, 3, 4, 5, 6);

    Flowable.defer(() -> {
        boolean[] skip = { false };
        return Flowables.<Pair>orderedMerge(
                a.<Pair>map(Left::new), b.<Pair>map(Right::new)
            )
            .distinctUntilChanged()
            .buffer(2, 1)
            .flatMapIterable(buf -> {
                if (skip[0]) {
                    skip[0] = false;
                    return Collections.emptyList();
                }
                if (buf.size() == 2) {
                    if (buf.get(0).value == buf.get(1).value) {
                        skip[0] = true;
                        return Collections.singletonList(new Both(buf.get(0).value));
                    }
                    return buf.subList(0, 1);
                }
                return buf;
            });
    })
    .subscribe(System.out::println);
}

where Flowables.orderedMerge在里面RxJava 2 扩展库.