В rxjs как я могу создать группы последовательных значений?

Question

У меня есть поток значений в этой форме:

[1,2,3,1,2,1,1,1,2...]

Я хотел бы преобразовать его в поток групп в виде:

[[1,2,3],[1,2],[1],[1],[1,2]...]

Новая группа должна создаваться каждый раз, когда значение становится равным 1.

Answer 1

Вы можете использовать оператор bufferWhen () для сбора излучаемых значений до тех пор, пока наблюдаемое не изменит значение.В этом примере я использую субъект для передачи мелкой копии потока в буфер.

Буфер будет излучать всякий раз, когда испускается число 1.Если поток начинается с 1, то создается пустой массив.Так что я отфильтровал это.

const {from, Subject} = rxjs;
const {filter, bufferWhen, tap, skip} = rxjs.operators;

const stream = from([1,2,3,1,2,1,1,1,2]);
const trigger = new Subject();

stream.pipe(
   tap(v => trigger.next(v)),
   bufferWhen(() => trigger.pipe(filter(v => v === 1))),
   filter(v => v.length)
).subscribe(x => console.log(x));

<script src="https://unpkg.com/@reactivex/rxjs@6.x/dist/global/rxjs.umd.js"></script>

Вы можете использовать scan () , если хотите запустить триггер, когда значение уменьшается отпредыдущая.Что может быть немного лучше по логике, но использование 1 в качестве триггера соответствует вопросу.

Answer 2

Для протокола, я выкладываю решение, альтернативное принятому ответу.

ПРИМЕЧАНИЕ: для конечных потоков concat(of(1)) требуется до scan(), для distinctUntilChanged для испускания последней наблюдаемой.

const {of, from, ReplaySubject} = rxjs;
const {map, concat, scan, distinctUntilChanged, concatAll, toArray, delay} = rxjs.operators;

//considering infinite stream
const stream = from([1,2,3,1,2,1,1,1,2]);
stream.pipe(
    scan( (acc, val) => {
        if(val===1){
            acc.complete()
            acc=new ReplaySubject();
        }
        acc.next(val);
        return acc;
    }, new ReplaySubject()), 
    distinctUntilChanged(), 
    map(toArray()), 
    concatAll() );

Было бы неплохо собрать отзывы о предпочтительном решении.