RXJS：交替组合流的元素答案

【问题标题】：RXJS: alternately combine elements of streamsRXJS：交替组合流的元素
【发布时间】：2015-02-22 10:51:58
【问题描述】：

我想交替组合多个流的元素：

var print = console.log.bind(console);

var s1 = Rx.Observable.fromArray([1, 1, 5]);
var s2 = Rx.Observable.fromArray([2, 9]);
var s3 = Rx.Observable.fromArray([3, 4, 6, 7, 8]);

alternate(s1, s2, s3).subscribe(print); // 1, 2, 3, 1, 9, 4, 5, 6, 7, 8

alternate的函数定义如何？

【问题讨论】：

标签： rxjs

【解决方案1】：

在处理从数组（如您的示例中）创建的可观察对象时使用 zip 和 concatMap，或者在处理本质上异步的可观察对象时使用 zip 和 flatMap。

Rx.Observable
  .zip(s1, s2, s3, function(x,y,z) { return [x,y,z]; })
  .concatMap(function (list) { return Rx.Observable.from(list); })
  .subscribe(print); // 1, 2, 3, 1, 9, 4

请注意，一旦源 observables 之一完成，这将不再继续。那是因为zip 是严格“平衡的”，它一直等到所有源都有匹配的事件。在处理完整的源代码时，您需要的是一个有点松散的 zip 版本。

【讨论】：

【解决方案2】：

如果源 observables 没有发出一个值（例如 undefined），则此解决方案有效：

var concat = Rx.Observable.concat;
var repeat = Rx.Observable.repeat;
var zipArray = Rx.Observable.zipArray;
var fromArray = Rx.Observable.fromArray;

var print = console.log.bind(console);

var s1 = fromArray([1, 1, 5]);
var s2 = fromArray([2, 9]);
var s3 = fromArray([3, 4, 6, 7, 8]);

alternate(s1, s2, s3).subscribe(print);

function alternate() {
    var sources = Array.slice(arguments).map(function(s) {
        return concat(s, repeat(undefined))
    });
    return zipArray(sources)
            .map(function(values) {
                return values.filter(function(x) {
                    return x !== undefined;
                });
            }).takeWhile(function(values) {
                return values.length > 0;
            }).concatMap(function (list) { return fromArray(list); })
}

ES6 中的相同示例：

const {concat, repeat, zipArray, fromArray} = Rx.Observable;

var print = console.log.bind(console);

var s1 = fromArray([1, 1, 5]);
var s2 = fromArray([2, 9]);
var s3 = fromArray([3, 4, 6, 7, 8]);

alternate(s1, s2, s3).subscribe(print);

function alternate(...sources) {
    return zipArray(sources.map( (s) => concat(s, repeat(undefined)) ))
            .map((values) => values.filter( (x) => x !== undefined ))
            .takeWhile( (values) => values.length > 0)
            .concatMap( (list) => fromArray(list) )
}

【讨论】：