Business
Jobs
  • About Us
  • Solutions
    • Job Postings
      Post your job and receive qualified candidates in 48h.
    • Candidate Assessments
      500+ technical and psychological tests, plus anti-fraud.
    • Headhunting
      Tailor-made executive search from start to finish.
    • Payroll + EOR
      Payroll dispersal and EOR across 15+ LATAM countries.
  • Pricing
  • Jobs

0

255
Views
RxJS sequential delay between emissions

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  
about 4 years ago · Juan Pablo Isaza
3 answers
Answer question

0

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))
);
about 4 years ago · Juan Pablo Isaza Report

0

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.

about 4 years ago · Juan Pablo Isaza Report

0

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

about 4 years ago · Juan Pablo Isaza Report
Answer question
Find remote jobs

Discover the new way to find a job!

Top jobs
Top job categories
Business
Post vacancy Pricing Sales
Legal
Terms and conditions Privacy policy
© 2026 PeakU Inc. All Rights Reserved.
Andres GPT
Show me some job opportunities
There's an error!