Rxjs自定义waitFor操作符实现用户未确认时忽略事件问题求解

原有代码问题 #

• pipe中添加的take(1)会让interval流第一次触发后直接结束，后续事件永远不会触发
• 自定义waitFor操作符逻辑不符合需求，你需要的是跳过等待状态下的所有新事件，而非等待状态恢复后补发之前的事件

最佳实现方案（推荐使用exhaustMap） #

exhaustMap是RxJS专为这类场景设计的操作符：上游触发新事件时，如果内部Observable还未完成，就直接忽略新事件，完美匹配「弹窗未处理完就忽略新触发事件」的需求，不需要手动维护状态变量。
完整代码如下：

import { Component, OnInit } from '@angular/core';
import { interval, of, from } from 'rxjs';
import { exhaustMap, tap } from 'rxjs/operators';
import { confirm } from 'devextreme/ui/dialog';

@Component({
  selector: 'my-app',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css'],
})
export class AppComponent implements OnInit {
  name = 'Angular';

  ngOnInit(): void {
    this.demo();
  }

  demo = () => {
    const obs$ = interval(5000);
    obs$.pipe(
      exhaustMap(() => this.handleMessage())
    ).subscribe();
  };

  // 处理逻辑返回Observable即可，exhaustMap会自动等待它完成
  handleMessage() {
    // alert示例用下面的代码，of包裹同步逻辑返回Observable
    alert('Hello');
    return of(null);

    // 实际DevExtreme confirm场景用下面的代码，把异步逻辑转成Observable
    /*
    return from(confirm(
      'Data has been updated from another tab. Would you like to reload it?',
      'Warning'
    )).pipe(
      tap(result => {
        if (result) window.location.reload();
      })
    )
    */
  }
}

保留手动状态控制的兼容实现 #

如果需要保留原有BehaviorSubject做状态管理的逻辑，可以直接用filter操作符过滤状态，不需要自定义操作符：

demo = () => {
  const obs$ = interval(5000);
  obs$.pipe(
    filter(() => this.instance.value.status)
  ).subscribe(this.handleMessage.bind(this));
};

内容的提问来源于stack exchange，提问作者Udemytur