Поток массивов в массив операторов потоков

Question

В Rx.js, как превратить поток массивов в массив потоков, например, у меня есть поток следующих: ['0a', '0b'], ['1a' «1b»], [ «2а», «2b», «2c»] и я хочу, чтобы получить следующие потоки:

0a---1a---2a---> 
0b---1b---2b---> 
      2c---> 

существуют ли какие-либо операторы делать что-то подобное, или я должен написать один из царапина?

Answer 1

Что-то, как это должно работать

stream. 
    flatMap(array => 
    Rx.Observable.from(
     array.map((obj, i) => {index: i, ...obj}) 
    ) 
).groupBy(x => x.index,). 
    subscribe(x => 
    x.map((x,i) => subscribe(x)) 
) 

Answer 2

Вы можете легко достичь этого с помощью существующих операторов.

То, что вы хотите достичь, это очень похоже на то, что описано здесь: RXJS: alternately combine elements of streams

Он предлагает два способа:

1. с помощью оператора Rx.Observable.zip (принимает в качестве аргумента массив наблюдаемых и испускать поток элементов, элемент которого при индексе n представляет собой x-мерное значение, излучаемое n-м наблюдаемым)

   Это решение, применяемое в вашем примере, остановится на 1a,1b, потому что итоговое обследование rvable будет завершен, как только одно из наблюдаемых завершено.

2. расширяя свои массивы, чтобы дать им все ту же длину, выполняя фиктивными значениями, и применяя Rx.Observable.zip оператор

В те оба варианта:

• если вы удалите последнюю строку , .concatMap.... вы получите поток массива, такой как [0a,0b], [1a,1b], [2a,2b,2c], из которого вы можете легко составить карту по индексу (.map(function(array){return array[N];}) доставит вам [Na,Nb...]), чтобы получить поток, который вы хотите.
• ИЛИ вы можете сохранить тот же код и добавить .filter(function(value,index){return index % N == I}), где N является количеством потоков, и I это поток вы хотите, то есть поток со значениями (Ia,Ib...)

документация по zip оператора>http://reactivex.io/documentation/operators/zip.htmlhttps://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/zip.md,