javascript - rxjs 链可以并行触发，但内部逻辑应该是同步的（.next 和 concat）

Question

在我的案例中，字段可以最终确定（与字段相关的操作已执行）。完成后，我需要更新 2 个列表：

• alreadyExecutedFields: string[] --> 普通数组
• remainingFieldsToExecute: BehaviorSubject<string[]> --> 行为主体，因为 .next 需要触发其他逻辑。这个逻辑可以并行触发，但我想防止这种情况发生，因为这个逻辑中有一个拼接可能会表现得不正确。（拼接一个索引，在平行链中被移除。）

因此，当需要完成某个字段时，我会调用：

this.finalize$.next(field);

finalize$ 链看起来像这样：

this.finalize$.pipe(
        concatMap((field: string) => {
            return new Promise<void>((resolve) => {
                console.log('START', field);
                this.alreadyExecutedFields.push(field);
                const remainingFieldsToExecute = this.remainingFieldsToExecute$.value;
                remainingFieldsToExecute.splice(remainingFieldsToExecute.indexOf(field), 1);
                this.remainingFieldsToExecute$.next(remainingFieldsToExecute);
                console.log('END', field);
                resolve();
            });
        }),
    ).subscribe(() => { });

但是由于某种原因，当 2 个 finalize$.next 调用紧接着发生时， concatMap 不会等待前一个的承诺。此外，当我尝试在 END 日志和解析周围设置超时时，它不会等待先前的解析。

在我的情况下起作用的不是使用 concatMap，而是使用带有标志的 setInterval，它锁定了正在更新列表的代码部分。

但是如何以更好的方式做到这一点呢？或者以正确的方式使用或不使用 concat 管道。

Answer 1

1. 修改alreadyExecutedFields您可以使用tap用于副作用的运算符
2. 从remainingFieldsToExecute您那里提取价值可以使用withLatestFrom
3. 然后修改remainingFieldsToExecute- 你可以再次使用tap

  alreadyExecutedFields: string[] = [];
  remainingFieldsToExecute$: BehaviorSubject<string[]> = new BehaviorSubject<
    string[]
  >(['1', '2', '3', '4', '5']);
  finalize$ = new BehaviorSubject('1');

  ngOnInit() {
    this.finalize$
      .pipe(
        tap(field => this.alreadyExecutedFields.push(field)),
        withLatestFrom(this.remainingFieldsToExecute$),
        tap(([field, remainings]) => {
          remainings.splice(remainings.indexOf(field), 1);
          this.remainingFieldsToExecute$.next(remainings);
        })
      )
      .subscribe(([_, data]) => {
        console.group();
        console.log('*** NEW EMIT ***');
        console.log('current field:', _);
        console.log('remainingFieldsToExecute', data);
        console.log('alreadyExecutedFields:', this.alreadyExecutedFields);
        console.groupEnd();
      });

    this.finalize$.next('2');
    this.finalize$.next('3');
  }

演示：https ://stackblitz.com/edit/angular-ivy-ensh7w?file=src/app/app.component.ts

日志：

可以简化为：

    this.finalize$
      .pipe(
        mergeMap(field =>
          this.remainingFieldsToExecute$.pipe(
            map(remFields => {
              this.alreadyExecutedFields.push(field);
              remFields.splice(remFields.indexOf(field), 1);
              return remFields;
            })
          )
        )
      )
      .subscribe()