rx-java - 接收 : a zip-like operator that continues after one of the streams ended?

Question

我正在寻找组合异步开始和结束的流(可观察量):

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

我需要它的用途:将音频流添加在一起。它们是音频“ block ”流，但我将在这里用整数表示它们。现在正在播放第一个剪辑:

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

然后第二个开始，稍后:

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

将它们相加的结果应该是:

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

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

注意:我正在使用 RxPy，但这里的社区似乎很小，而且 Rx 运算符似乎跨语言相当通用，所以我也将其标记为 rx-java 和 rx-js。

Answer 1

我会通过将其分成两部分来解决这个问题。首先，我想要一些需要 Observable<Observable<T>> 的东西并产生 Observable<Observable<T>[]>其中数组仅包含“事件”(即不完整)的可观察值。每当将新元素添加到外部可观察量时，以及每当内部可观察量之一完成时，都会发出一个包含适当可观察量的新数组。这本质上是对主流的“扫描”缩减。

一旦你有了可以做到这一点的东西，你就可以使用 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))
);

我假设零个事件流不应产生任何内容，一个事件流应产生其自己的元素，两个或多个流应全部压缩在一起(所有这些都反射(reflect)在 map 应用程序中)。

我的(诚然相当有限的)测试使这种组合产生了您想要的结果。

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