RxJs如何將兩個重疊的可觀察到的合併成一個

Question

我有兩個可觀察：

-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15-| 
-13--14--15--16--17--18--19-----20---------21--------------22------23--24--> 

第一個包含一些越來越多，但經過一段時間後停止（這些是來自數據庫的光標結果） 第二類是不斷涌現越來越多的人首先包含一些數字，但不要停止發射。 （這些都是新插入的數據到數據庫）

我想這兩個可觀察一下一個連續的觀察到這樣的：

-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15-16-17-18-19-20-21-----22------23--24--> 

此觀察到的包含了每一個號碼只有一次，保持順序發光。

如何使用盡可能少的內存來解決問題？

Answer 1

我認爲最好的辦法在這裏是緩衝b $，直到$流達到b $，然後發出所有b $的緩衝項並切換到b $。事情是這樣的：

const a = '-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15'; 
 
const b = '-13--14--15--16--17--18--19-----20---------21--------------22------23--24'; 
 

 
const fromMarble = str => Rx.Observable.from(str.split('-')).concatMap(x => Rx.Observable.of(x).delay(1)).filter(v => v.length).map(x => parseInt(x)); 
 

 
const a$ = fromMarble(a).share(); 
 
const b$ = fromMarble(b).share(); 
 

 
const switchingSignal$ = Rx.Observable.combineLatest(a$, b$.take(1), (a, b) => a >= b).filter(x => x).take(1).share(); 
 

 
const distinct$ = Rx.Observable.merge(
 
\t a$.takeUntil(switchingSignal$).map(x => x + '(from a)'), 
 
\t b$.buffer(switchingSignal$).take(1).mergeMap(buffered => Rx.Observable.from(buffered)).map(x => x + '(from b$ buffer)'), 
 
\t b$.skipUntil(switchingSignal$).map(x => x + '(from b$)') 
 
); 
 

 
distinct$.subscribe(console.log);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.0/Rx.js"></script>

Answer 2

您可以通過採取從（.concat）與第二流連接起來的第一流的所有元素，除了最新的一個（.last）之前（.skipWhile含）的元素

const a = '-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15' 
 
const b = '-13--14--15--16--17--18--19-----20---------21--------------22------23--24' 
 
const fromMarble = str => Rx.Observable.defer(() => { 
 
    console.log('side effect from subscribing to: ' + str); 
 
    return Rx.Observable.from(str.split('-').filter(v => v.length)); 
 
}); 
 

 
const a$ = fromMarble(a); 
 
const b$ = fromMarble(b); 
 

 
const distinct$ = Rx.Observable.concat(
 
    a$, 
 
    a$.last().switchMap(latest => 
 
    // .skipWhile + .skip(1) => skipWhile but inclusive 
 
    b$.skipWhile(v => v !== latest).skip(1) 
 
), 
 
); 
 

 
distinct$.subscribe(e => console.log(e));

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

做到這一點另外，如果訂閱時有副作用（例如訂閱時 - 將創建新光標），則可以通過使用例如const a$ = fromMarble(a).shareReaplay()來爲所有訂戶共享該副作用。

你可以閱讀更多有關共享的副作用：老文檔中

• 爲RxJS V4 - 4.8 Use the publish operator to share side-effects
• 並從這篇文章 - RxSwift: share vs replay vs shareReplay