兩個有序可觀測物的全外連接

Question

假設我們有兩個可觀測物Observable<Integer> o1和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] ] 

的加入應該是高效的和非常大的或無限流工作。

該解決方案在拉方案中很容易。有沒有一種慣用的方法來實現這一點？

Answer 1

有這個沒有單一的運營商，但它可以從標準和擴展運營商組成的行爲：

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); 
} 

其中Flowables.orderedMerge在RxJava 2 Extensions library。