javascript - 当其中没有重叠的异步操作时完成的异步操作的可观察对象(主题)

Question

我希望答案是forkJoin/Promises.all，但有点多，请耐心等待。

我有一个 promises 的来源，它可以随机到达，我需要一些方式来说“当到目前为止到达的所有 promise 都完成时，让我知道”。

在基于 Promise 的解决方案中，我最初考虑使用 Promise.all 但 promises 可能仍然“到达”而其他人尚未完成。有趣的是，在 https://stackoverflow.com/a/37819138/239168 上有一个“可迭代的 Promise.all”的巧妙解决方法。

我正在尝试以 Rx 方式执行此操作。稍微阅读文档后，我认为 forkJoin 是 Promise.all 的等价物，但是，同样的问题，没有时间点可以安全地调用 forkJoin 或 Promise.all 因为总是可以添加一个，而另一个仍在等待中......因为我现在可能没有任何意义，我想我会要求一些指导。

设置

(如果这很傻，请保持笑声，我是 Rx 的新手......)

我有一个 Subject，我想知道它里面的所有 promises 什么时候完成......而且它总是可以随时获得新添加的 promises......

private promiseSource = new Subject<Promise<any>>();
promises$ = this.promiseSource.asObservable();

每当一个新的 promise “到来”时，我只是将它添加到主题中

this.promiseSource.next(somePromise);

我想奇迹般地发生的是 - 只要主题只包含已完成的 promise ，就让主题“完成”。

例如

promises$.magicFlatMapForkJoinConcatMapTrickery().subscribe({
  next: x => ...,
  error: err => ...,
  complete: () => {
     console.log('all promises we got so far are done');
     // nice to have, I want this to keep "listening" for new promises
     promiseSource.youAreNotREALYCompletePleaseReset(); 
  }
});

或者换句话说，我有一个异步操作的可观察对象，如果我们看一下内容，我们可以看到重叠的异步操作，我想知道什么时候没有重叠，例如

|<-async action 1->|   |<-async action 3->|
          |<-async action 2->|                     |<-async action 4->|

                                          /\      /\
                                        find this gap

如果这些是例如 http 调用，我基本上是在问 - 当没有打开的 http 调用时告诉我。

tl;dr

如何在 RxJS 世界中实现这个基于 Promises 的答案...

https://stackoverflow.com/a/37819138/239168

Answer 1

如果我正确地解释了您的问题，那么您只对指示是否存在未决 promise 的信号感兴趣。

使用 merge 和 scan 来创建一个发出待处理 promises 计数的可观察对象非常容易，从那里，您应该能够创建您想要的任何信号喜欢。

基本上，每次主题发出一个 promise 时，待处理 promise 的数量应该增加。每次这些 promise 中的一个解决时，计数就会减少。

const promises = new Rx.Subject();

const pendingCount = Rx.Observable
  .merge(
    promises.mapTo(1),
    promises.mergeMap(p => Rx.Observable.from(p).mapTo(-1))
  )
  .scan((acc, value) => acc + value, 0)
  .do(count => console.log(`${count} pending promise(s)`));

const doneSignal = pendingCount
  .filter(count => count === 0)
  .mapTo("done");

doneSignal.subscribe(signal => console.log(signal));

const timeoutPromise = (delay) => new Promise(resolve => setTimeout(resolve, delay));

promises.next(timeoutPromise(200));
setTimeout(() => promises.next(timeoutPromise(200)), 100);
setTimeout(() => promises.next(timeoutPromise(200)), 300);
setTimeout(() => promises.next(timeoutPromise(200)), 700);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>