I have a Observable observableA. How can I create another Observable observableB which emits value from observableA, but the time between each emission is at least 1000 milliseconds?
For example:
observableA: X...500ms..Y..1500ms..Z...600ms..T
observableB: X..1000ms..Y..1500ms..Z..1000ms..T
You can always use a silent timer as a lower bound for each new emission length. Then concatMap will handle back pressure for you.
const bound = timer(1000).pipe(ignoreElements());
const observableB = observableA.pipe(
concatMap(v => merge(of(v), bound))
);
You could use interval and zip to create an observable that fires 1000ms after every value on observableA, and then zip it with observableA -
const observableA = create(subscriber => {
window.setTimeout(() => {
subscriber.next(1);
window.setTimeout(() => {
subscriber.next(2);
window.setTimeout(() => {
subscriber.next(3);
subscriber.complete();
}, 500);
}, 1500);
}, 600);
});
const delay = concat(from([undefined]), observableA).pipe(
flatMap(() => interval(1000).take(1))
);
const observableB = zip(observableA, delay).map(([first]) => first);
observableB.subscribe((val) => {
console.log(JSON.stringify(val) + " " + Date.now());
});
If you don't want to delay the first value, you can instead do
const delay = concat(from([undefined]), observableA.pipe(
flatMap(() => interval(1000).take(1))
));
To add an immediate first value.
If you want to put it into a single chain you could use the following using connect() operator:
const obsA$ = range(5).pipe(
concatMap((v) => of(v).pipe(delay(Math.random() * 2000)))
);
const MIN_DELAY = 1000;
let observerReceivedTime = new Date().getTime();
obsA$
.pipe(
startWith(null),
connect((shared$) => {
let lastEmissionTime = null;
return shared$.pipe(
concatMap((value) => {
let delayedValue$;
if (!lastEmissionTime) {
// Don't delay the first emission.
delayedValue$ = of(value);
} else {
const delayBetweenValues = new Date().getTime() - lastEmissionTime;
// Wait for at least `MIN_DELAY`.
delayedValue$ =
delayBetweenValues > MIN_DELAY
? of(value)
: of(value).pipe(delay(MIN_DELAY - delayBetweenValues));
}
return delayedValue$.pipe(
// Remember time when the last value was reemited.
tap(() => (lastEmissionTime = new Date().getTime())),
);
})
);
})
)
.subscribe((v) => {
const now = new Date().getTime();
console.log(
`observer receives emission with at least ${MIN_DELAY}ms delay:`,
now - observerReceivedTime,
'value:',
v
);
observerReceivedTime = now;
});
Live demo: https://stackblitz.com/edit/rxjs-wmfetc?devtoolsheight=60&file=index.ts
I'm using RxJS 7 connect() operator to create a scoped variable lastEmissionTime so I don't need to make any side-effects (though the chain would be a little shorter and easier to understand to be honest).
Basically, it's just measuring time between the previous emission and the current one and will adjust delay() to always be at least MIN_DELAY.
In the observer callback I'm just measuring timestamps to prove that the delay is always at least 1000ms as you wanted.
Edit: This is an alternative using mergeScan which is better in my opinion:
obsA$
.pipe(
startWith(null),
mergeScan(
([_, lastEmissionTime], currValue) => {
let delayedValue$;
if (!lastEmissionTime) {
// Don't delay the first emission.
delayedValue$ = of(currValue);
} else {
const delayBetweenValues = new Date().getTime() - lastEmissionTime;
// Wait for at least `MIN_DELAY`.
delayedValue$ =
delayBetweenValues > MIN_DELAY
? of(currValue)
: of(currValue).pipe(delay(MIN_DELAY - delayBetweenValues));
}
return delayedValue$.pipe(
// Remember time when the last value was reemited.
map((value) => [value, new Date().getTime()])
);
},
[null, null],
1 // We need to set concurrency to `1`.
),
map(([value]) => value)
)
Live demo: https://stackblitz.com/edit/rxjs-taeyzx?devtoolsheight=60&file=index.ts