javascript - 我怎样才能将一个可观察对象变成一个由长轮询可观察对象组成的可观察对象，这些可观察对象在特定值上完成？

Question

我正在使用 RxJs 创建一个交互式网页。

这是我想要实现的:

我有一个生成 token 的应用程序。这些 token 可以由外部实体使用。

当用户创建 token 时，页面开始轮询网络服务器以了解其状态(已使用或未使用)。当 token 被消耗时，页面刷新。

因此，当创建 token 时，每 2 秒向服务器发送一个请求，询问 token 是否已被使用。

我有一个 Observable 字符串，代表我的 generatedTokens。

我实际上已经有了一个使用 Rx.Scheduler.default 类的工作实现，它允许我手动执行操作。但是，我不禁觉得应该有一个更简单、更优雅的解决方案。

这是当前代码:

class TokenStore {
  constructor(tokenService, scheduler) {
    // actual implementation omitted for clarity
    this.generatedTokens = Rx.Observable.just(["token1", "token2"]);

    this.consumedTokens = this.generatedTokens
      .flatMap(token => 
        Rx.Observable.create(function(observer) {
          var notify = function() {
            observer.onNext(token);
            observer.onCompleted();
          };
          var poll = function() {
            scheduler.scheduleWithRelative(2000, function() {
                // tokenService.isTokenConsumed returns a promise that resolves with a boolean
                tokenService.isTokenConsumed(token)
                  .then(isConsumed => isConsumed ? notify() : poll());
              }
            );
          };
          poll();
        }));
  }
}

是否有类似“repeatUntil”的方法？我正在寻找一个与上面的代码做同样事情的实现，但看起来更像这样:

class TokenStore {
  constructor(tokenService, scheduler) {
    // actual implementation omitted for clarity
    this.generatedTokens = Rx.Observable.just(["token1", "token2"]);

    this.consumedTokens = this.generatedTokens
      .flatMap(token =>
        Rx.Observable.fromPromise(tokenService.isTokenConsumed(token))
                     .delay(2000, scheduler)
                      // is this possible?
                     .repeatUntil(isConsumed => isConsumed === true));
  }
} 

Answer 1

有趣的是，在发布问题几分钟后，答案让我印象深刻。我想 rubberducking 可能并没有那么傻。

无论如何，答案由两部分组成:

• repeatUntil 可以通过 repeat()、filter() 和 first() 的组合来实现

• fromPromise 有一些内部惰性缓存机制，导致后续订阅不会触发新的 AJAX 请求。因此我不得不重新使用 Rx.Observable.create

解决方法:

class TokenStore {
  constructor(tokenService, scheduler) {
    // actual implementation omitted for clarity
    this.generatedTokens = Rx.Observable.just(["token1", "token2"]);

    this.consumedTokens = this.generatedTokens
      .flatMap(token =>
        // must use defer otherwise it doesnt retrigger call upon subscription
        Rx.Observable
        .defer(() => tokenService.isTokenConsumed(token))
        .delay(2000, scheduler)
        .repeat()
        .filter(isConsumed => isConsumed === true)
        .first())
    .share();
  }
} 

一个小旁注:“share()”确保两个 observable 都是热的，这避免了每个订阅者都会导致 ajax 请求开始触发的情况。