Reactive Programmingでこのコードを書くにはどうすればいいですか？

Question

私はちょうど反応型プログラミングを使いこなし始めましたが、コードを書くだけで十分ですが、私が期待したものが得られないときに何が起こっているのか分かりません。私は実際にブログの投稿以外のメンターを利用することはできません。私は自分が持っている状況に非常に良い解決策を見出すことはできませんでした。私は正しいアプローチに興味があります。

問題：

私は部分的にバーオブジェクトの配列で構成されているのFooを取得する必要があります。私は、WebサービスからBarオブジェクトを取得します。そこで私は各WebサービスコールをIObservableとして表現しました。そこから、完成前に0または1の要素が必要です。

• 各IObservableインスタンスをサブスクライブする必要があります。
• 最大2秒のタイムアウトまで待ちます。
• 両方の配列の完全またはタイムアウトのいずれかが起こる：
  • が生成されたバーオブジェクトのいずれかを有する配列を作成
  、そのバー[]を使用してFooのオブジェクトを生成し（0であってもよいです）。

私は、ソートのコードのこのビットでこれを達成：多くの理由のために、右のようではありません

public Foo CreateFoo() { 

     var producer1 = webService.BarGenerator() 
            .Timeout(TimeSpan.FromSeconds(2), Observable.Empty<Bar>()); 
     var producer2 = // similar to above 

     var pipe = producer1.Concat(producer2); 

     Bar[] result = pipe.ToEnumerable().ToArray(); 

     ... 
} 

。最も明白なのは、Concat（）はシーケンスをパラレルではなく連続的に開始するため、これは4秒のタイムアウトです。私はそれがブロックされていることに本当に気をつけませんが、実際にはそれが働いているアーキテクチャにとっては便利です。私は、このメソッドはIObservableの発電機となって微細なんだけど、私がしようとすると、それが難しくするように見えるここにいくつかの余分な注意点があります：

1. 私はその順序でproducer1とproducer2の結果を置くために、最終的な配列を必要とします両方とも結果が得られた場合
2. 私はTestSchedulerを使用してタイムアウトを確認したいが、それでも成功していない、私は明らかにスケジューラを全く理解していない。
3. これは最終的にはプルモデルで、Fooが明確な点でそれを必要とするものは何でも、それを「オンザフライで」受信する価値はありません。たぶんこれは「Rxを使わない」という答えを傾けるかもしれない。正直言って、私はタスクベースのAPIに切り替えるだけで十分に立ち往生しています。しかし、私はRxでこれにどのようにアプローチするのかを見たいと思っています。私は学びたいからです。

Answer 1

var pipe = producer1 
     .Merge(producer2) 
     .Buffer(Observable.Timer(TimeSpan.FromSeconds(2), testScheduler)) 
     .Take(1); 

    var subscription = pipe 
     .Select(list => new Foo(list.ToArray()) 
     .Subscribe(foo => {} /* Do whatever you want with your foo here.*/); 

Buffer（2秒で私たちの場合）ウィンドウの中に放出されたすべての要素を取得し、リストを出力します。

あなたの代わりにあなたができるサブスクリプションの、あなたのプル・モデルに固執する場合：

var list = await pipe; 
var foo = new Foo(list.ToArray()); 

//....