According to the documentation, the buffer transform will wait for a delay before emitting any values. What I'd like is to get the current value immediately, then only update every X seconds.
I've not been able to achieve this with rxjs yet. The closest I've come is to bind the observable then use a setTimeout function to rebind after the buffer timeout occurs. This has a side effect of clearing the current value for those X seconds before emitting the current values.
Any ideas?
Thanks!
Assuming by "current value immediately" you mean "first value as soon as it emits", you can buffer on the second element to the last, and merge in the first:
// source$: Observable<T>
const pub_source$ = source$.publish();
Observable.merge(
pub_source$.take(1).map(first => [first]),
pub_source$.skip(1).buffer(Observable.interval(X))
);
pub_source$.connect();
The source needs to be cold so that take(1) and skip(1) relate to the same element, so we use publish. The first element is also wrapped to keep the output type T[] consistent.
There's an easier way using the zip operator, see lightbulb note in learnrxjs
Combined with interval or timer, zip can be used to time output from another source!
// Useful for slow source that emits at around the same rate as interval
// but suffers back-pressure with fast-emitting source
const interval = 1000
const output = Observable.zip(source, Observable.timer(0, interval))
.map(x => x[0])
The Observable.timer 'regulates' the output from source. Note, timer's first parameter sets the delay for the first emit.
Working example: CodePen
Footnote
I just realized this will create back-pressure (build-up of un-emitted values) if you have a lot of events per second, so buffer is the better way to go with a fast emitting source.
// Buffered version for fast source
const output2 = source.buffer(Observable.timer(0, interval))
.filter(x => x.length) // filter out empty buffer emits
.flatMap(x => Observable.from(x)) // optional, converts array back to single emits
