I have below stream of data coming
1 > 4 > 0 > 3 > 1 > 0 > 0 > 0 > 1 > 2 > 0 > 0 > 0 > 0 > 0 > 0
I want 0 be emitted only if it repeated 3 times sequentially or more, so the expected result will be something like
1 > 4 > 3 > 1 > 0 > 1 > 2 > 0
How can I do that in rxjs?
You might want to implement it as in the following example:
const { from } = rxjs;
const { scan, filter, map} = rxjs.operators;
const input = [
1, 4, 0, 3, 1, 0, 0, 0,
1, 2, 0, 0, 0, 0, 0, 0,
2, 0, 2, 4, 0, 0, 3, 9,
4, 3, 0, 0, 0, 0, 0, 0,
];
from(input).pipe(
scan(({ counter }, current) => {
if (current === 0) return { current, counter: ++counter, emit: counter == 3 }
else return { current, counter: 0, emit: true };
}, { emit: false, counter: 0, current: undefined }),
filter(x => x.emit),
map(x => x.current)
)
.subscribe(console.log)<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.6.2/rxjs.umd.min.js"></script>Indeed, you could use scan/filter/map. Another option is to use your own custom operator for it.
const {
Observable,
from
} = rxjs;
function groupRepeatedNumbers(opts = {
minLimit: 3,
desiredNumber: 0
}) {
return function(source) {
let numberCounter = 0;
return new Observable(subscriber => {
source.subscribe({
next(value) {
if (value !== opts.desiredNumber) {
numberCounter = 0;
subscriber.next(value);
} else if (++numberCounter === opts.minLimit) {
subscriber.next(value);
}
},
error(error) {
subscriber.error(error);
},
complete() {
subscriber.complete();
}
})
});
}
}
const input = [
1, 4, 0, 3, 1, 0, 0, 0,
1, 2, 0, 0, 0, 0, 0, 0,
2, 0, 2, 4, 0, 0, 3, 9,
4, 3, 0, 0, 0, 0, 0, 0,
];
from(input).pipe(
groupRepeatedNumbers()
)
.subscribe(console.log)<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.6.2/rxjs.umd.min.js"></script>You can see that you can even configure your operator to act as you want, like:
groupRepeatedNumbers({
minLimit: 5,
desiredNumber: 3
}) // Will group the number 3, and only emit when it gets 5 or more occasions
You can use defer to keep track of a variable (like a counter) in a custom operator while providing each subscriber with its own new variable (which is important if you have multiple subscribers).
import { defer, MonoTypeOperatorFunction, Observable } from 'rxjs';
import { filter } from 'rxjs/operators';
function sequentialLimit<T>(
condition: (value: T) => boolean,
limit: number
): MonoTypeOperatorFunction<T> {
return (source: Observable<T>) => defer(() => {
let counter = 0;
return source.pipe(
filter(v => {
condition(v) ? counter++ : counter = 0;
return counter === 0 || counter === limit;
})
);
});
}
Usage:
const input = [
1, 4, 0, 3, 1, 0, 0, 0, 1, 2, 0, 0, 0, 0, 0, 0,
2, 0, 2, 4, 0, 0, 3, 9, 4, 3, 0, 0, 0, 0, 0, 0,
];
from(input).pipe(
sequentialLimit(v => v === 0, 3)
).subscribe(console.log)