我如何在 RxJs 中实现类似 "bufferWhile" 的东西？

Question

我使用 RxJs 处理从文件重放的数据。每个数据项包含一个timereceived 属性。重播时，我想创建缓冲区，其中包含最初在给定时间跨度 x 内收到的所有数据项。换句话说：我想将所有项目添加到当前缓冲区，而接收到的第一个缓冲区元素和接收到的当前元素之间的时间跨度小于时间跨度 x。

示例测试：

it('should be able to create buffers based on time received', function() {
    // given
    let source = require('rx').Observable.from([
        {val: 1, timereceived: 10},
        {val: 2, timereceived: 20},
        {val: 3, timereceived: 100},
        {val: 4, timereceived: 110},
        {val: 5, timereceived: 120}
    ]);

    // when
    let actual = app.bufferWithTimeReceived(source, 105).toArray();

    // then
    console.log(actual);
    assert.equal(actual.length, 2); // first contains val 1-3, second val 4-5
})

如果我不想重播文件中的所有数据而只是实时接收它，我可以为此使用 bufferWithTime 并且没问题。

用另一个例子更新

    // given
    let source = require('rxjs/Rx').Observable.from([
        {val: 1, timereceived: 10},
        {val: 2, timereceived: 20},
        {val: 3, timereceived: 100},
        {val: 4, timereceived: 110},
        {val: 5, timereceived: 120},
        {val: 6, timereceived: 9920},
        {val: 7, timereceived: 9930}
    ]);

    // when
    app.bufferWithTimeReceived(source, 30).subscribe(console.log);


    // then
    // expected output would be [val 1-2][val 3-5][val 6-7] (empty arrays in between would be ok)

更新结束

现在我尝试了不同的方法。我的最后一个是：

exports.bufferWithTimeReceived = (source, timespan) => {
    return Rx.Observable.defer(() => Rx.Observable.create(function (observer) {
        let currBuffer = [];
        source.subscribe(x => {
            if (currBuffer.length == 0)
                currBuffer = [x];
            else {
                if (x.timereceived-currBuffer[0].timereceived < timespan)
                    currBuffer.push(x);
                else {
                    observer.onNext(currBuffer);
                    currBuffer = [x];
                }
            }
        },
        (err)=>observer.onError(err),
        ()=>observer.onCompleted());
    }));
};

不幸的是，这只会导致 oArrayObservable { source: Defer { _f: [Function] } } 作为错误消息，这不是很有帮助。我还想知道对我有什么帮助？！

额外问题：有什么提示可以让这个缓冲区重叠吗？

Answer 1

我会这样做，但请注意，这是 RxJS 5。但是，mergeAll 也应该在 RxJS 4 中可用（可能名称不同）。

const THRESHOLD = 105;

const data = [
  {val: 1, timereceived: 10},
  {val: 2, timereceived: 20},
  {val: 3, timereceived: 100},
  {val: 4, timereceived: 110},
  {val: 5, timereceived: 120}
];
const source = Observable.from(data)
  .groupBy(item => parseInt(item.timereceived / THRESHOLD))
  .map(observable => observable.toArray())
  .mergeAll()
  .subscribe(console.log);

groupBy 运算符为 parseInt(item.timereceived / THRESHOLD) 返回的每个键创建一个新的 Observable。然后我用 toArray() 链接它，因为我想在重新发送它们之前收集它的所有项目，并且 mergeAll() 订阅所有 Observables（在本例中为 2）并重新发送它们的项目，每个项目始终是一个数组来源 Observable。

这会产生以下输出：

[ { val: 1, timereceived: 10 }, { val: 2, timereceived: 20 }, { val: 3, timereceived: 100 } ]
[ { val: 4, timereceived: 110 }, { val: 5, timereceived: 120 } ]

参见：

Answer 2

感谢马丁的回答，我再次考虑了 groupBy，这似乎对我有用：

exports.bufferWithTimeReceived = (source, timespan) => {
        let currTime;
        return source.groupBy(x => {
            if (!currTime)
                currTime = x.timereceived;
            if (x.timereceived-currTime > timespan)
                currTime = x.timereceived;
            return currTime;            
        })
        .map(observable => observable.toArray())
        .mergeAll()
};

但我想知道 bufferWhen 是否可以提供更优雅的解决方案？！

我如何在 RxJs 中实现类似 "bufferWhile" 的东西？

How do I implement something like "bufferWhile" in RxJs?

javascript

reactive-programming

rxjs

rxjs5

用另一个例子更新

更新结束