4

我正在寻找组合异步开始和结束的流（observables）：

-1----1----1----1---|->
     -2----2--|->
[ optional_zip(sum) ]
-1----3----3----1---|->

我需要它：将音频流添加在一起。它们是音频“块”流，但我将在这里用整数表示它们。所以播放了第一个剪辑：

-1----1----1----1---|->

然后第二个开始，稍晚一点：

     -2----2--|->

将它们按总和组合的结果应该是：

-1----3----3----1---|->

但是，如果任何压缩流结束，标准 zip 就会完成。即使其中一个流结束，我也希望这个 optional_zip 继续运行。有没有办法在 Rx 中做到这一点，或者我必须通过修改现有的 Zip 自己实现它？

注意：我使用的是 RxPy，但这里的社区似乎很小，而且 Rx 运算符似乎在各种语言中都很通用，所以我也将它标记为 rx-java 和 rx-js。

4

2 回答 2

2

我会通过将它分成两部分来解决这个问题。首先，我想要一个接受Observable<Observable<T>>并产生一个Observable<Observable<T>[]>数组只包含“活动”（即非完整）可观察对象的东西。每当一个新元素被添加到外部 observable 中，并且任何时候内部 observable 之一完成时，都会发出一个包含适当 observable 的新数组。这本质上是对主流的“扫描”缩减。

一旦你有了可以做到这一点的东西，你就可以使用 flatMapLatest 和 zip 来获得你想要的东西。

我在第一部分的基本尝试如下：

function active(ss$) {
    const activeStreams = new Rx.Subject();
    const elements = [];
    const subscriptions = [];

    ss$.subscribe(s => {
        var include = true;
        const subscription = s.subscribe(x => {}, x => {}, x => {
            include = false;
            const i = elements.indexOf(s);
            if (i > -1) {
                elements.splice(i, 1);
                activeStreams.onNext(elements.slice());
            }
        });

        if (include) {
            elements.push(s);
            subscriptions.push(subscription);
            activeStreams.onNext(elements.slice());
        }   
    });

    return Rx.Observable.using(        
        () => new Rx.Disposable(() => subscriptions.forEach(x => x.dispose())),
        () => activeStreams
    );
}

从那里，您只需将其拉上拉链并将其展平，如下所示：

const zipped = active(c$).flatMapLatest(x =>
    x.length === 0 ? Rx.Observable.never()
  : x.length === 1 ? x[0]
  : Rx.Observable.zip(x, (...args) => args.reduce((a, c) => a + c))
);

我假设零个活动流不应该产生任何东西，一个活动流应该产生自己的元素，两个或多个流应该全部压缩在一起（所有这些都反映在地图应用程序中）。

我的（诚然相当有限的）测试使这种组合产生了您所追求的结果。

顺便说一句，好问题。我还没有看到任何解决问题第一部分的方法（尽管我绝不是 Rx 专家；如果有人知道已经这样做的东西，请发布详细信息）。

于 2016-03-10T02:50:12.507 回答
1

所以我得到了一些我认为可以满足您大部分需求的代码。基本上，我创建了一个函数zipAndContinue，它将像 一样运行zip，但只要某些底层流仍有数据要发出，它就会继续发出项目。此功能仅 [简要] 使用冷可观察对象进行了测试。

此外，欢迎更正/增强/编辑。

function zipAndContinue() {
    // Augment each observable so it ends with null
    const observables = Array.prototype.slice.call(arguments, 0).map(x => endWithNull(x));
    const combined$ = Rx.Observable.combineLatest(observables);

    // The first item from the combined stream is our first 'zipped' item
    const first$ = combined$.first();

    // We calculate subsequent 'zipped' item by only grabbing
    // the items from the buffer that have all of the required updated
    // items (remember, combineLatest emits each time any of the streams
    // updates).
    const subsequent$ = combined$
        .skip(1)
        .bufferWithCount(arguments.length)
        .flatMap(zipped)
        .filter(xs => !xs.every(x => x === null));

    // We return the concatenation of these two streams
    return first$.concat(subsequent$)
}

以下是使用的实用函数：

function endWithNull(observable) {
    return Rx.Observable.create(observer => {
        return observable.subscribe({
            onNext: x => observer.onNext(x),
            onError: x => observer.onError(x),
            onCompleted: () => {
                observer.onNext(null);
                observer.onCompleted();
            }
        })
    })
}

function zipped(xs) {
    const nonNullCounts = xs.map(xs => xs.filter(x => x !== null).length);

    // The number of streams that are still emitting
    const stillEmitting = Math.max.apply(null, nonNullCounts);

    if (stillEmitting === 0) {
        return Rx.Observable.empty();
    }

    // Skip any intermittent results
    return Rx.Observable.from(xs).skip(stillEmitting - 1);
}

这是示例用法：

const one$ = Rx.Observable.from([1, 2, 3, 4, 5, 6]);
const two$ = Rx.Observable.from(['one']);
const three$ = Rx.Observable.from(['a', 'b']);

zipAndContinue(one$, two$, three$)
    .subscribe(x => console.log(x));

// >> [ 1, 'one', 'a' ]
// >> [ 2, null, 'b' ]
// >> [ 3, null, null ]
// >> [ 4, null, null ]
// >> [ 5, null, null ]
// >> [ 6, null, null ]

这是一个 js-fiddle（您可以单击运行，然后打开控制台）：https ://jsfiddle.net/ptx4g6wd/

于 2016-03-09T23:46:03.357 回答