javascript – RxJs将流分成多个流

--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...>

您可以使用窗口并共享源Observable. 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' ]
[ 'b','b','b' ]
[ 'c','c','c' ]
[ 'd','d','d' ]
[ 'e' ]

此解决方案的问题是订阅源的顺序.我们需要窗口通知程序在第一个bufferCount之前订阅.否则,首先进一步推送项目,然后检查它是否与前一个项目不同.filter(([oldValue,newValue])…).

这意味着需要在窗口之前将发射延迟一个(这是第一个.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,newValue]) => oldValue !== newvalue)
    )
    .concatMap(obs => obs.toArray())
    .subscribe(console.log);

connectable.connect();

输出是一样的.