RxJS：延迟传入的 observables答案

【问题标题】：RxJS: delaying incoming observablesRxJS：延迟传入的 observables
【发布时间】：2020-03-20 12:52:33
【问题描述】：

我有一个 SnackbarComponent，它负责为用户显示一个带有信息的 Snackbar。通知来自我在SnackbarComponent 中订阅的 NotificationService。传入的消息被推送到消息数组并在 3 秒后从它转移。在我的 html 模板中，我遍历消息并显示它们。

export class SnackbarComponent implements OnInit, OnDestroy {
  private notificationSubscription: Subscription;
  messages: NotificationMessage[] = [];

  constructor(private notificationService: NotificationService) {
  }

  ngOnInit() {
    this.notificationSubscription = this.notificationService.notification$
      .subscribe(
        (message: NotificationMessage) => {
          this.messages.push(message);
          setTimeout(() => {
            this.messages.shift();
          }, 3000);
        }
      )
  }

  ngOnDestroy(): void {
    this.notificationSubscription.unsubscribe();
  }

问题是，我想一次只显示 3 条消息并延迟其他消息（如果当前少于 3 条消息，则立即显示）。

我有一个可行的解决方案，如下所示：

private notificationSubscription: Subscription;
messages: NotificationMessage[] = [];
messagesQueue: NotificationMessage[] = [];


  ngOnInit() {
    this.notificationSubscription = this.notificationService.notification$
      .subscribe(
        (message: NotificationMessage) => {
          if (this.messages.length < 3) {
            this.messages.push(message);
            this.timeoutMessage();
          } else {
            this.messagesQueue.push(message);
          }
        }
      )
  }

  private timeoutMessage() {
    setTimeout(() => {
      this.messages.shift();
      if (this.messagesQueue.length > 0) {
        this.messages.push(this.messagesQueue.shift());
        this.timeoutMessage();
      }
    }, 3000);
  }

如果有 3 个或消息，则将新消息推送到 messagesQueue 中，并将 messagesQueue 中的 timeoutMessage() 对象转移到消息数组中。

但我觉得这个解决方案并不优雅。有没有一种 RxJS 方法可以通过一些 pipe 运算符以某种方式延迟消息？

【问题讨论】：

标签： javascript angular typescript rxjs observable

【解决方案1】：

您可以使用 delay 运算符或条件延迟 delayWhen 根据条件进行延迟

import {delay} from 'rxjs/operators';

this.notificationService.notification$.pipe(
  delay(3000)
).subscribe();

delayWhen

示例

this.notificationService.notification$.pipe(
  delayWhen(message => this.messages.length < 3 ? interval(0) : interval(3000))
).subscribe();

【讨论】：

它会延迟每个通知。如果消息数组长度小于 3，我不想延迟它们。并考虑这种情况：3 个通知同时到来，2500 毫秒后第四个通知到来。将显示 3 条通知，在 3000 毫秒后它们应该消失，第 4 条应在 3000 毫秒处再显示 3000 毫秒，但第 4 条将在 5500 毫秒处显示
在这里查看条件延迟rxjs.dev/api/operators/delayWhen
必须以某种方式计算间隔时间，将其设置为 3000 会导致顺序问题，请查看：stackblitz.com/edit/angular-feyqox

【解决方案2】：

您可以使用BufferCount 运算符来做到这一点。它一次会发出 3 个值，我添加了第二个延迟以确保。

import {bufferCount,delay } from 'rxjs/operators';

this.notificationService.notification$.pipe(
  bufferCount(3),delay(3000)
).subscribe();

这是一个堆栈闪电战example：

【讨论】：

如果只有一个通知来了，它不会显示，直到 2 来。
@mlemmy ，它会显示。您可以尝试我共享的 stackblitz 链接。尝试将可观察值更改为 1 并检查。它会发出它。 BufferCount 只是对值进行分组。
请看：stackblitz.com/edit/angular-169vb1 它显示了我对这个解决方案有什么问题。另外我不喜欢它强迫我在订阅中有一个数组