RxJs將流拆分爲多個流

Question

如何基於分組方法將永不結束的流拆分爲多個結束流？

--a--a-a-a-a-b---b-b--b-c-c---c-c-d-d-d-e...> 

到這些觀測

--a--a-a-a-a-| 
      b---b-b--b-| 
         c-c---c-c-| 
            d-d-d-| 
             e...> 

正如你所看到的，a是在開始的時候，我收到b後，我將不再獲得a所以它應該結束。這就是爲什麼正常的groupBy不好。

Answer 1

您可以使用window和share源觀測。還有一個小竅門與bufferCount(2, 1)：

const str = 'a-a-a-a-a-b-b-b-b-c-c-c-c-d-d-d-e'; 
const source = Observable.from(str.split('-'), Rx.Scheduler.async).share(); 

source 
    .bufferCount(2, 1) // delay emission by one item 
    .map(arr => arr[0]) 
    .window(source 
     .bufferCount(2, 1) // keep the previous and current item 
     .filter(([oldValue, newValue]) => oldValue !== newValue) 
    ) 
    .concatMap(obs => obs.toArray()) 
    .subscribe(console.log); 

這版畫（因爲toArray()）：

[ 'a', 'a', 'a', 'a', 'a' ] 
[ 'b', 'b', 'b', 'b' ] 
[ 'c', 'c', 'c', 'c' ] 
[ 'd', 'd', 'd' ] 
[ 'e' ] 

這種解決方案的問題是訂閱到source順序。我們需要window通知者在第一個bufferCount之前訂閱。否則，首先將物品推進一次，然後檢查物品是否與.filter(([oldValue, newValue]) ...)之前的物品不同。

這意味着將需要一個window之前推遲發射（這是第.bufferCount(2, 1).map(arr => arr[0])

或者，也許它更容易與publish()控制訂閱我的順序：

const str = 'a-a-a-a-a-b-b-b-b-c-c-c-c-d-d-d-e'; 
const source = Observable.from(str.split('-'), Rx.Scheduler.async).share(); 

const connectable = source.publish(); 

connectable 
    .window(source 
     .bufferCount(2, 1) // keep the previous and current item 
     .filter(([oldValue, newValue]) => oldValue !== newValue) 
    ) 
    .concatMap(obs => obs.toArray()) 
    .subscribe(console.log); 

connectable.connect(); 

輸出是相同

Answer 2

也許有人能想出更簡單的東西，但這個工程（小提琴：https://fiddle.jshell.net/uk01njgc/）...

let counter = 0; 

let items = Rx.Observable.interval(1000) 
.map(value => Math.floor(value/3)) 
.publish(); 

let distinct = items.distinctUntilChanged() 
.publish(); 

distinct 
.map(value => { 
    return items 
    .startWith(value) 
    .takeUntil(distinct); 
}) 
.subscribe(obs => { 
    let obsIndex = counter++; 
    console.log('New observable'); 
    obs.subscribe(
    value => { 
     console.log(obsIndex.toString() + ': ' + value.toString()); 
    }, 
    err => console.log(err), 
    () => console.log('Completed observable') 
); 
}); 

distinct.connect(); 
items.connect(); 

Answer 3

這裏有一個變化，包裝所有的suscription分享給你...

const stream = ...; 

// an Observable<Observable<T>> 
// each inner observable completes when the value changes 
const split = Observable 
    .create(o => { 
    const connected = stream.publish(); 

    // signals each time the values change (ignore the initial value) 
    const newWindowSignal = connected.distinctUntilChanged().skip(1); 

    // send the observables to our observer 
    connected.window(newWindowSignal).subscribe(o); 

    // now "start" 
    return connected.connect(); 
    });