CODEWITHSUNDEEP
rxjsreactive-streams
Marinos An
Marinos An

Reputation: 10808

In rxjs how can I create groups of sequential values?

I have a stream of values in this form:

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

I would like to convert it to a stream of groups in the form:

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

New group should be created every time value becomes 1.

Upvotes: 4

Views: 484

Answers (2)

Marinos An
Marinos An

Reputation: 10808

For the record, I'm posting a solution alternative to the accepted answer.

NOTE: For finite streams concat(of(1)) is required before scan(), for distinctUntilChanged to emit the last observable.

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

It would be nice to collect some feedback regarding preferred solution.

Upvotes: 0

Reactgular
Reactgular

Reputation: 54751

You can use the bufferWhen() operator to collect emitted values until an observable emits a value. In this example I use a subject to emit a shallow copy of the stream to the buffer.

The buffer will emit when ever the number 1 is emitted. If the stream starts with 1 then an empty array is emitted. So I filter that away.

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/[email protected]/dist/global/rxjs.umd.js"></script>

You can use the scan() if you wanted to emit the trigger when the value decreases from the previous. Which is maybe a little better in logic, but using 1 as the trigger fits with the question.

Upvotes: 6

Related Questions