asynchronous - 基于异步事件管理订阅

Question

我正在寻找一种方法来简化这一点并避免在我的管道内管理订阅。

关于我的代码的总体思路：

this.start$ubject // waiting for users call to start
  .pipe(
    mergeMap(() => from(this.thirdPartyService.start())), // now i need to wait for smth to start
    tap(() => {
      // only after thirdPartyService started i can subscribe because prior to that they are undefined
      this.subscriptions.push(
        this.thirdPartyService.alfa$.subscribe(this.first$ubject),
        this.thirdPartyService.beta$.subscribe(this.second$ubject),
      );
    }),
  );

有什么办法可以处理吗？类似的东西，takeWhile但订阅？

标签： asynchronousrxjsrxjs5rxjs6

Answer 1

试试这样：

// waits for start$ubject, then waits for thirdPartyService, then starts.
this.subscriptions.push(
  this.start$ubject.pipe(
    switchMap(() => from(this.thirdPartyService.start())), // waiting start
    switchMap(() => merge( // listening on both of the streams
      this.thirdPartyService.alfa$.pipe(
        tap(this.first$ubject),
      ),
      this.thirdPartyService.beta$.pipe(
        tap(this.second$ubject),
      ),
    )),
  ).subscribe(),
);

// waits for start$ubject or for thirdPartyService, then starts.
this.subscriptions.push(
  merge(this.start$ubject, from(this.thirdPartyService.start()).pipe(
    switchMap(() => merge( // listening on both of the streams
      this.thirdPartyService.alfa$.pipe(
        tap(this.first$ubject),
      ),
      this.thirdPartyService.beta$.pipe(
        tap(this.second$ubject),
      ),
    )),
  ).subscribe(),
);