Empresas
Empleos
  • Sobre nosotros
  • Soluciones
    • Publicación de vacantes
      Publica tu vacante y recibe candidatos calificados en 48h.
    • Evaluación de candidatos
      500+ pruebas técnicas y psicológicas, más anti-fraude.
    • Headhunting
      Búsqueda ejecutiva a la medida de principio a fin.
    • Nómina + EOR
      Dispersión de nómina y EOR en más de 15 países de LATAM.
  • Precios
  • Empleos

0

257
Vistas
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 Respuestas
Responde la pregunta

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 Denunciar

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 Denunciar

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 Denunciar
Responde la pregunta
Encuentra empleos remotos

¡Descubre la nueva forma de encontrar empleo!

Top de empleos
Top categorías de empleo
Empresas
Publicar vacante Precios Comercial
Legal
Términos y condiciones Política de privacidad
© 2026 PeakU Inc. All Rights Reserved.
Andres GPT
Recomiéndame algunas ofertas
Necesito ayuda