Empresas
Empregos
  • Sobre nós
  • Soluções
    • Publicação de vagas
      Publique sua vaga e receba candidatos qualificados em 48h.
    • Avaliações de candidatos
      Mais de 500 testes técnicos e psicológicos, mais anti-fraude.
    • Headhunting
      Busca executiva personalizada do início ao fim.
    • Folha de Pagamento + EOR
      Dispersão de folha e EOR em mais de 15 países da LATAM.
  • Preços
  • Empregos

0

256
Visualizações
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 Respostas
Responde à pergunta

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 Relatório

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 Relatório

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 Relatório
Responde à pergunta
Encontrar trabalhos remotos

Descubra a nova forma de encontrar um emprego!

melhores empregos
Principais categorias de trabalho
Empresas
Postar vaga Preços Comercial
Jurídico
Termos e Condições Política de privacidade
© 2026 PeakU Inc. All Rights Reserved.
Andres GPT
Recomende algumas ofertas para mim
Preciso de ajuda