发射之间的 RxJS 顺序延迟

Question

我有一个 Observable observableA。如何创建另一个 Observable observableB 从 observableA 发射值，但每次发射之间的时间至少为 1000 毫秒？
例如：

observableA:  X...500ms..Y..1500ms..Z...600ms..T  
observableB:  X..1000ms..Y..1500ms..Z..1000ms..T  

Answer 1

您可以使用 interval 和 zip 创建一个 observable，在 observableA 上的每个值之后触发 1000 毫秒，然后使用 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());
});

如果你不想延迟第一个值，你可以改为这样做

const delay = concat(from([undefined]), observableA.pipe(
  flatMap(() => interval(1000).take(1)) 
));

添加一个直接的第一个值。

Answer 2

如果您想将其放入单个链中，您可以使用 connect() 运算符：

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;
  });

现场演示：https://stackblitz.com/edit/rxjs-wmfetc?devtoolsheight=60&file=index.ts

我正在使用 RxJS 7 connect() 运算符来创建一个作用域变量 lastEmissionTime，所以我不需要产生任何副作用（尽管说实话链会更短一些并且更容易理解） .

基本上，它只是测量上一次发射和当前发射之间的时间，并将delay() 调整为始终至少为MIN_DELAY。

在观察者回调中，我只是测量时间戳以证明延迟始终至少为 1000 毫秒，如您所愿。

编辑：这是使用mergeScan 的替代方法，我认为这更好：

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)
  )

现场演示：https://stackblitz.com/edit/rxjs-taeyzx?devtoolsheight=60&file=index.ts

Answer 3

您始终可以使用静默计时器作为每个新发射长度的下限。然后 concatMap 将为您处理背压。

const bound = timer(1000).pipe(filter(_ => false));
const observableB = observableA.pipe(
  concatMap(v => merge(of(v), bound))
);