«Распространение» последовательность среди других последовательностей

Question

У меня есть IObservable, который производит последовательность так:

A1, A2, A3, A4, B1, B2, B3, B4, C1, C2, C3, C4, D1, D2, D3, D4, ... 

Я хотел бы иметь метод:

IObservable<T>[] Distribute(IObservable<T>, count) 

То, при вызове на IObservable, будет производить другие IObservables - четыре в этом случае, так что каждый из них будет представлять собой элементы из относительной «позиции» в исходной последовательности (псевдокод):

GetOutput(unzipped[0]) = { A1, B1, C1, D1, ... }; 
GetOutput(unzipped[1]) = { A2, B2, C2, D2, ... }; 

Это идентично человеку, открывающему карты для N человек (N = count), когда вы даете одну карту для каждого человека в последовательности, а затем начинаете все заново.

Вопросов:

1. Можно ли получить эту функциональность с существующими методами Rx?
2. Если нет, как я могу реализовать его сам? Я чувствую, что с помощью Where с делением по модулю индекса может быть приемлемым, за исключением, если ответ на вопрос 1 выше «да» ...

Answer 1

Когда в Rx, вы как правило, остаются в IObservable земле до тех пор, возможное. Так естественнее подпись будет IObservable<IObservable<T>> Distribute(IObservabe<T> source, int count), который будет выглядеть следующим образом:

public static IObservable<IObservable<T>> Distribute<T>(this IObservable<T> source, int count) 
{ 
    var toReturn = source.Select((t, i) => Tuple.Create(t, i)) 
     .GroupBy(t => t.Item2 % count) 
     .Select(g => g.Select(t => t.Item1)) 
    return toReturn; 
} 

Вы можете получить его как массив, как это: