javascript - Rx - 根据内容拆分 Observable(分组依据直到更改)

Question

让我先给出一些问题的背景。目标是使用 Rx 来处理带有分页的简单搜索屏幕。

在输入端，用户可以添加各种过滤条件，还可以更改当前显示的页面(如果结果集较大)。为简单起见，这可以用 {filter, page} 对流建模。

在输出端，需要将结果数据以表格形式展示，需要更新页数。这意味着 {data, count} 对流。

连接两者的逻辑如下:每当过滤器参数发生变化时，向服务器发送“计数”查询以获取页面计数。每当过滤器参数或页面索引发生变化时，向服务器发送“选择”查询以获取结果数据。等到两者都可用再更新屏幕。

这是一个以最小的方式模拟这个的 plunk。 s1/s2/s3 表示过滤器参数，p1/p2/p3 是页面索引，#s1/#s2/#s3 是过滤器的“计数”响应，而像“s2 p3”这样的东西是一个“选择”响应给定的过滤器和页面组合。您可以看到各种失败的方法，已注释掉。

http://plnkr.co/xurmUO

似乎解决方案应该使用 groupBy 或 window 之类的东西，但两者都无法产生预期的结果。使用 groupBy，您可以做的是从输入流创建组，并根据过滤器参数将输入项分配给组。然后，您仅在创建新组时触发“计数”查询，并为组中的每个单独项目触发“选择”查询。

这样做的问题是组一直持续到源流的末尾，因此如果您重复之前出现的过滤器参数，则不会创建新组。因此，也不会发生“计数”查询。 groupByUntil 运算符允许指定到期持续时间，这几乎 解决了所有问题。然后您可以创建一个新组(由于输入中的不同过滤器参数)触发当前组的到期。像这样:

inputs
  .groupByUntil(
    function(input) { return input.filter; },
    null,
    function(inputGroup) { 
      var filter = inputGroup.key;
      return inputs.firstOrDefault(function (input) {
        return input.filter !== filter;
      });
    }
  );

不幸的是，这似乎比应有的时间晚触发，具体取决于运算符执行的顺序。结果，组边界偏离了一个，您最终会进行额外的“计数”调用。我还尝试使用窗口运算符，它有一个类似的问题:一个额外的项目在关闭之前进入窗口。

我想要的是 groupBy 的一种变体，它一次只有一个事件组，当收到带有新 key 的项目时，使前一个过期。您可以通过某种方式组合现有的运算符来构建它吗？或者您是否必须返回到 Observable.create，并使用低级操作来执行此操作？

编辑:作为引用，这是我使用自定义运算符的解决方案，通过 Observable.create 完成。

Rx.Observable.prototype.splitBy = function(keySelector) {
  var source = this;
  var currentGroup = null;
  var lastKey = null;
  return Rx.Observable.create(function(observer) {
    return source.subscribe(
      function(value) {
        var key = keySelector(value);
        // Create a new group if the key changed (or this is the first value)
        if (currentGroup === null || key !== lastKey) {
          // Close previous group
          if (currentGroup !== null) {
            currentGroup.onCompleted();
          }
          lastKey = key;
          // Emit current group
          currentGroup = new Rx.Subject();
          currentGroup.key = key;
          observer.onNext(currentGroup);
        }
        currentGroup.onNext(value);
      },
      // Forward errors/completion to current group and the main stream
      function(error) {
        if (currentGroup !== null) {
          currentGroup.onError(error);
        }
        observer.onError(error);
      },
      function() {
        if (currentGroup !== null) {
          currentGroup.onCompleted();
        }
        observer.onCompleted();
      });
  });
};

Answer 1

由于您已经发现的原因，groupBy 运算符系列不起作用。你想要 flatMapLatest:

var count = inputs.flatMapLatest(function (input) {
    // some code that returns an
    // Rx observable that will eventually
    // return the count
    return rxQueryForCount(input.filter);
});
count.subscribe(...);

你会写一些类似的东西来返回数据页面。

编辑:如果你想同步你的 2 个查询的结果，那么你可以这样做:

var count = inputs
    .distinctUntilChanged(null, function (a, b) {
        // however you decide if your filters have changed...
        return a.filter === b.filter;
    })
    .flatMapLatest(function (input) {
        // some code that returns an
        // Rx observable that will eventually
        // return the count
        return rxQueryForCount(input.filter)
            .startWith(undefined);
    });
var data = inputs
    .flatMapLatest(function (input) {
        return rxQueryForData(input.filter, input.index)
            .startWith(undefined);
    });
var results = Rx.Observable
    .combineLatest(count, data, function (c, d) {
        if (c === undefined || d === undefined) {
            return undefined;
        }

        return { count: c, data: d };
    });

每当查询重新运行时，结果都将是undefined，并且只会在两个查询都更新结果时才取值。您可以在您的 UI 逻辑中使用它，或者甚至只是添加一个 where 子句来过滤掉 undefined 值。