让我先说一下这个问题的背景.目标是使用Rx处理带分页的简单搜索屏幕.

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

在输出端,结果数据需要显示在表中,并且需要更新页数.这意味着{data,count}对的流.

连接两者的逻辑如下：每当过滤器参数改变时,向服务器发送“计数”查询以获取页面计数.每当过滤器参数OR页面索引发生变化时,向服务器发送“select”查询以获取结果数据.在更新屏幕之前,请等待两者都可用.

这是一个以最小的方式模拟这个的插件. s1 / s2 / s3表示过滤器参数,p1 / p2 / p3是页面索引,#s1 /#s2 / #s3是过滤器的’count’响应,类似’s2 p3’是’select’的响应给定过滤器和页面组合.注释掉后,您可以看到各种失败的方法.

http://plnkr.co/xurmUO

似乎解决方案应该使用像groupBy或window这样的东西,但是它们都不会产生预期的结果.使用groupBy,您可以执行的操作是从输入流创建组,并根据filter参数将输入项分配给组.然后,您只在创建新组时触发“计数”查询,并触发对组中每个项目的“选择”查询.

这样做的问题是组会持续到源流结束,因此如果重复先前发生的过滤器参数,则不会创建新组.因此,也不会发生“计数”查询. 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的一个变体,它一次只有一个活动组,当前一个活动组收到一个带有新密钥的项目时,它会过期.你能以某种方式结合现有的操作符来构建这个吗？或者您是否必须返回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();
      });
  });
};

解决方法:

由于您已经发现的原因,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 };
    });

每当查询重新运行时,结果将不确定,并且只有在两个查询都有更新结果时才会获取值.您可以在UI逻辑中使用它,甚至只需添加where子句来过滤掉未定义的值.