如何创建一个返回 Observable 的方法，该方法发出需要一个接一个执行的 2 个 Promise 的结果？

Question

我问了一个问题

Is Observable from chained promises equivalent of observables created with from and chained with concatMap?

在完全错误的前提下。看来我的两个解决方案都与我的意图无关。

我创建了一个返回 Observable 的方法，并调用了两个返回 Promise 的方法。我尝试了两种方法：

  public setItemInfos(itemInfos: IItemInfo[]): Observable<number> {
    return from(this.db.selectionItemInfos.clear().then(() => {
      return this.db.selectionItemInfos.bulkAdd(itemInfos);
    }));
  }

  public setItemInfos(itemInfos: IItemInfo[]): Observable<number> {
    const clear$ = from(this.db.selectionItemInfos.clear());
    const bulkAdd$ = from(this.db.selectionItemInfos.bulkAdd(itemInfos));

    return clear$.pipe(concatMap(() => bulkAdd$))
  }

用途将是：

myService.setItemInfos(itemInfos).subsribe(count => {
  console.log(`Cleared the table 1st and then added ${count} new items`);
});

我从两个版本中都认为：

1. table clear 是在 bulkAdd 开始时执行完成
2. bulkAdd 完成后，我从订阅中获取计数

这应该怎么做？或者可以做到吗？

Answer 1

这是（据我所知），我会怎么做。

一般来说，defer（或任何高阶运算符）是从 Promise 创建 observable 的更好方法。 Defer 让您可以将 Promise 的热切求值语义转化为可观察对象的惰性求值语义。

然后所有常用的可观察运算符等都会按预期运行。

public setItemInfos(itemInfos: IItemInfo[]): Observable<number> {
  const clear$ = defer(() => this.db.selectionItemInfos.clear());
  const bulkAdd$ = defer(() => this.db.selectionItemInfos.bulkAdd(itemInfos));

  return concat(clear$, bulkAdd$);
}

---

更新 1：

所以我想我可能知道你在追求什么。这并不是真正地道的 RxJS，因为它是声明式、命令式代码风格的交错组合。即便如此，这应该有效吗？我还没有完全测试它，但是有些修修补补，我认为这应该可以满足您的需求。

肯定有更好的方法来完成同样的事情，但是如果没有看到你所追求的大局，很难说。

interface Tagged<T> {
  payload: T,
  tag: number
}

class abitraryClass{

  private setItemInfoSub: Subject<Tagged<IItemInfo[]>>;
  private processItemInfo: Observable<Tagged<number>>;
  private itemInfoTag = 0;

  constructor(){
    this.setItemInfoSub = new Subject<Tagged<IItemInfo[]>>();
    this.processItemInfo = this.setItemInfoSub.pipe(
      concatMap(({tag, payload: itemInfos}) => this.db.selectionItemInfos.clear().pipe(
        ignoreElements(),
        concatWith(defer(() => this.db.selectionItemInfos.bulkAdd(itemInfos))),
        map(response => ({
          payload: response,
          tag
        }))
      )),
      shareReplay(1)
    );
    // Make the processing pipeline live at all times.
    this.processItemInfo.subscribe();
  }

  public setItemInfos(itemInfos: IItemInfo[]): Observable<number> {
    const myTag = this.itemInfoTag++;

    this.setItemInfoSub.next({
      payload: itemInfos,
      tag: myTag
    });

    return this.processItemInfo.pipe(
      filter(({tag}) => tag == myTag),
      map(({payload}) => payload)
    );
  }
}