如何在没有等待的情况下使用 observable

Question

我有以下代码，这似乎有效，但我很确定有一种更简洁的方法可以使用链接运算符来实现它，事实是我在上面挣扎了 2 个小时，但我无法得到它以另一种方式工作。

也许你能帮帮我？

  public synchronizeWithRemote$(state: Partial<StorageCustomerCase>, pdfDocument?: InkPDFDocument): Observable<ISynchronizationProgress> {
return new Observable((subscriber) => {
  // PDF as changed, we update the hash
  if (pdfDocument) {
    this.pdfInvalidationHash = nanoid();
    // Start synchronization task
    pdfDocument
      .serialize()
      .then((serializedDoc) => {
        const task = this.storageRef.put(serializedDoc, { contentType: "application/pdf" });

        // Listen for changes
        task.on("state_changed", ({ bytesTransferred, totalBytes }) => {
          subscriber.next({ bytesTransferred, totalBytes, percent: Math.round((bytesTransferred * 100) / totalBytes) });
        });

        return task; // Await the load to finish before uploading to firestore the metadatas
      })
      .then(() =>
        this.documentRef.update({ ...omit(state, "id"), pdf_invalidation_hash: this.pdfInvalidationHash } as StorageCustomerCase),
      )
      .then(() => subscriber.complete());
  } else {
    subscriber.next({ bytesTransferred: 0, percent: 100, totalBytes: 0 });
    // TODO: Update this function to use only observable by chaining
    // Just update doc
    this.documentRef
      .update({ ...omit(state, "id"), pdf_invalidation_hash: this.pdfInvalidationHash } as StorageCustomerCase)
      .then(() => {
        // Notify firebase of changes, once PDF has been successfully uploaded.
        subscriber.complete();
      });
  }
});

}

如果有PDF文档，应该发送进度指示器，上传后与firestore同步，否则，仅与firestore同步后完成。

问候， 安德烈亚斯

Answer 1

这应该与你对 Promise 的语义大致相同，只使用 Observables。由于显而易见的原因，我实际上无法为您测试这个，所以将其用作答案的草图而不是整个答案。

我也不确定fromEventPattern 是否适用于您的情况。看起来应该。

public synchronizeWithRemote$(state: Partial<StorageCustomerCase>, pdfDocument?: InkPDFDocument): Observable<ISynchronizationProgress> {
  
  const finish$ = defer(() => this.documentRef.update({
    ...omit(state, "id"), 
    pdf_invalidation_hash: this.pdfInvalidationHash 
  } as StorageCustomerCase));

  if(doc == null){
    // No updates, finish right away
    return finish$.pipe(
      startWith({ bytesTransferred: 0, percent: 100, totalBytes: 0 })
    );
  }
  
  // Stream of state Changes
  const update$ = defer(() => doc.serialize()).pipe(
    tap(_ => this.pdfInvalidationHash = nanoid()),
    map(serializedDoc => 
      this.storageRef.put(serializedDoc, { 
        contentType: "application/pdf" 
      })
    ),
    switchMap(task =>
      fromEventPattern(
        handler => task.on("state_changed", handler)
      ).pipe(
        map(({ bytesTransferred, totalBytes }) => ({ 
          bytesTransferred, 
          totalBytes, 
          percent: Math.round((bytesTransferred * 100) / totalBytes) 
        }))
      )
    )
  );
  
  // Wait for update$ to complete before subscribing to finish$
  return concat(update$, finish$);
}

将 promise 转换为 observables

许多 RxJS 操作符，如果给定一个 promise 将隐式地将它们转换为 obervable。以下是一些：

1. from(promise)
2. defer(() => promise)
3. mergeMap/switchMap/concatMap(val => promise)

我不是#1 的忠实粉丝。虽然 #2 和 #3 保持承诺的行为类似于 observable（在您订阅之前什么都不会发生），但 #1 将立即启动承诺。任何事情都有时间和地点，但由于使用 #1 引起的错误在这里相当普遍，所以我认为值得一提。

例子;如果我在这里使用#1 而不是#2 作为finish$，那么pdf_invalidation_hash 将具有错误的值。因为它的值取决于何时订阅 finish$（update$ 运行后，this.pdfInvalidationHash 应该更新）。