RXJS:交替组合流的元素 [英] RXJS: alternately combine elements of streams

问题描述

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

I'd like to alternately combine elements of multiple streams:

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的函数定义?

推荐答案

如果有一个值(例如 undefined)不是由源 observables 发出的，这个解决方案有效:

If there is a value (for example undefined) that is not emitted by the source observables, this solution works:

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 中的相同示例:

Same example in 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) )
}

这篇关于RXJS:交替组合流的元素的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！