rxjs - 丢弃未来的返回可观察量

Question

所以这是我的可观察代码:

  var suggestions =
        Rx.Observable.fromEvent(textInput, 'keyup')
          .pluck('target','value')
          .filter( (text) => {
              text = text.trim();
              if (!text.length) // empty input field
              {
                  this.username_validation_display("empty");
              }
              else if (!/^\w{1,20}$/.test(text))
              {
                  this.username_validation_display("invalid");
                  return false;
              }
              return text.length > 0;
          })
          .debounceTime(300)
          .distinctUntilChanged()
          .switchMap(term => {
              return $.ajax({
                type: "post",
                url: "src/php/search.php",
                data: {
                  username: term,
                  type: "username"
                }
              }).promise();
            }
          );
  suggestions.subscribe(
    (r) =>
    {
        let j = JSON.parse(r);
        if (j.length)
        {
            this.username_validation_display("taken");
        }
        else
        {
            this.username_validation_display("valid");
        }
    },
    function (e)
    {
        console.log(e);
    }
  );

我遇到的问题是当输入为空时我有另一段代码基本上返回'错误:空输入'但它被返回的observable覆盖.所以我想知道是否有到不顾一切的观测,如果一个方法text.length是0,而且当文本长度不为零重新订阅.

我已经考虑过,unsubscribe但不知道在哪里适合尝试.

Answer 1

也许这会起作用？我们的想法是在switchmap中加入空输入,以便关闭当前的observable,即在这种情况下也将其丢弃.

  var suggestions =
        Rx.Observable.fromEvent(textInput, 'keyup')
          .pluck('target','value')
          .filter( (text) => {
              text = text.trim();
              if (!text.length) // empty input field
              {
                  this.username_validation_display("empty");
              }
              else if (!/^\w{1,20}$/.test(text))
              {
                  this.username_validation_display("invalid");
                  return false;
              }
          })
          .debounceTime(300)
          .distinctUntilChanged()
          .switchMap(text=> {
              if (text.length > 0) {
              return $.ajax({
                type: "post",
                url: "src/php/search.php",
                data: {
                  username: text,
                  type: "username"
                }
              }).promise()}
              return Rx.Observable.empty();
            }
          );

Answer 2

我建议debounceTime在文本输入流之后移动调用,并在去抖动流上进行过滤.这样一个有效但过时的值不会潜入ajax调用.此外,当一个新的,可能无效的输入发生时,我们可能需要"取消"一个ajax调用(或者更准确地说,忽略它的响应),因为我们不希望覆盖正在进行的异步的"无效"状态.呼叫.takeUntil可以帮助你.另请注意,Rx.Observable.ajax()可以代替使用$.ajax().

顺便说一句,拥有一个专用的"验证状态"流可能很方便,因此可以this.username_validation_display(validationStatus)在订阅中的单个位置完成.

我想它是这样的:

const term$ = Rx.Observable.fromEvent(textInput, "keyup")
    .pluck("target", "value")
    .map(text => text.trim())
    .distinctUntilChanged()
    .share();

const suggestion$ = term$
    .debounceTime(300)
    .filter(term => getValidationError(term).length === 0)
    .switchMap(term => Rx.Observable
        .ajax({
            method: "POST",
            url: "src/php/search.php",
            body: {
                username: term,
                type: "username"
            }
        })
        .takeUntil(term$)
    )
    .map(result => JSON.parse(result));

const validationStatus$ = Rx.Observable.merge(
    term$.map(getValidationError),
    suggestion$.map(getSuggestionStatus));

// "empty", "invalid", "taken", "valid" or ""
validationStatus$
    .subscribe(validationStatus => this.username_validation_display(validationStatus));

// Helper functions
function getValidationError(term) {
    if (term.length === 0) {
        return "empty";
    }

    if (!/^\w{1,20}$/.test(term)) {
        return "invalid";
    }

    return "";
}

function getSuggestionStatus(suggestion) {
    return suggestion.length > 0 ? "taken" : "valid";
}

上述代码的行为在某种意义上也是不同的,即在用户键入有效输入的时间点和建议从服务器到达的时间点之间,username_validation_display将具有空值,即除了"空"之外,"无效","采用"和"有效"状态也会有一个中间空状态,可以用UI中的某些进度指示替换.

更新:我还可以考虑一种应该具有等效行为的替代方法,但相应的代码可能看起来更简单:

const username$ = Rx.Observable.fromEvent(textInput, "keyup")
    .pluck("target", "value")
    .map(text => text.trim())
    .distinctUntilChanged();

const validationStatus$ = username$.switchMap(username => {
    const validationError = getValidationError(username);
    return validationError ?
        Rx.Observable.of(validationError) :
        Rx.Observable.timer(300)
            .switchMapTo(getAvailabilityStatus$(username))
            .startWith("pending");
});

// "invalid", "empty", "taken", "valid" or "pending"
validationStatus$.subscribe(
    validationStatus => this.username_validation_display(validationStatus));

// Helper functions
function getAvailabilityStatus$(username) {
    return Rx.Observable
        .ajax({
            method: "POST",
            url: "src/php/search.php",
            body: {
                username: username,
                type: "username"
            }
        })
        .map(result => JSON.parse(result))
        .map(suggestions => suggestions.length > 0 ? "taken" : "valid");
}

function getValidationError(username) {
    if (username.length === 0) {
        return "empty";
    }

    if (!/^\w{1,20}$/.test(username)) {
        return "invalid";
    }

    return null;
}

这样我们可以同步检查每个不同的输入.如果输入为空或无效(通过getValidationError()调用确定),我们会立即发出验证错误.否则,我们立即发出"挂起"状态并启动一个计时器,该计时器将在300毫秒后产生异步用户名可用性检查.如果在计时器过去之前到达新文本或从服务器到达可用性状态,我们取消内部流并重新开始验证,这要归功于switchMap.所以它应该和它一样工作debounceTime.