使用RX同步多個事件

Question

假設我有三個產品在列表中。爲了實現某種行動，所有這三者都需要具有某種特定類型。爲了找出產品的類型，我需要撥打服務電話並等待迴應。

我想要做的就是等待所有三個響應（也許在發生錯誤時使用超時），並且收集所有信息時，決定是否啓用可能的操作。

我曾經通過一些計數器或重置事件來跟蹤已完成的事件來解決此問題，但我想看看是否可以使用Rx以更清晰的方式執行此操作。

由於我不太熟悉Rx，我正在尋找一些提示/指針。我明白我可以使用

Observable.FromEventPattern 

因爲我等待的事件。我訂閱並等待迴應並處理它。我只是不清楚如何組合多個事件。

Answer 1

的Combinator的您正在尋找的是CombineLatest

假設你有一個這樣的類：

public class Foo 
{ 
    public delegate void FooEventHandler(object sender, EventArgs args); 

    public event FooEventHandler FirstEvent = delegate {};  
    public event FooEventHandler SecondEvent = delegate {};  
    public event FooEventHandler ThirdEvent = delegate {};  

    public void DoIt() 
    { 
     FireOne(); 
     FireTwo(); 
     FireThree(); 
    } 

    public void FireOne() 
    { 
     Console.WriteLine("Firing event 1..."); 
     Thread.Sleep(1000); 
     FirstEvent(this, new EventArgs()); 
    } 
    public void FireTwo() 
    { 
     Console.WriteLine("Firing event 2..."); 
     Thread.Sleep(1000); 
     SecondEvent(this, new EventArgs()); 
    } 
    public void FireThree() 
    { 
     Console.WriteLine("Firing event 3..."); 
     Thread.Sleep(1000); 
     ThirdEvent(this, new EventArgs()); 
    } 
} 

首先你要「轉換」這些事件來Observable：

var foo = new Foo(); 
var firstWatcher = Observable.FromEventPattern(foo, "FirstEvent"); 
var secondWatcher = Observable.FromEventPattern(foo, "SecondEvent"); 
var thirdWatcher = Observable.FromEventPattern(foo, "ThirdEvent"); 

現在你會想要「所有這些都已經發射的時候只有發射」選擇器，它是CombineLatest：

var allDone = Observable.CombineLatest(firstWatcher, secondWatcher, thirdWatcher); 

並對其進行測試：

using(allDone.Subscribe(_ => Console.WriteLine("Boop! You sunk my battleship!"))) 
{ 
    foo.DoIt(); 
}  

另類 「測試工具」：

var foo = new Foo(); 
var firstWatcher = Observable.FromEventPattern(foo, "FirstEvent"); 
var secondWatcher = Observable.FromEventPattern(foo, "SecondEvent"); 
var thirdWatcher = Observable.FromEventPattern(foo, "ThirdEvent"); 

var allDone = Observable.CombineLatest(firstWatcher, secondWatcher, thirdWatcher); 

// keep a handle on the subscription    
IDisposable subscription = null; 

// to prevent premature exiting... 
var blocker = new ManualResetEvent(false); 

// explicit subscribe 
subscription = allDone.Subscribe(
    whoCares => 
    { 
     Console.WriteLine("BOOM! We're done!"); 
     // always clean up after yourself 
     if(subscription != null) 
     { 
      subscription.Dispose(); 
     } 
     // it's ok, we can quit now 
     blocker.Set(); 
    }); 

foo.DoIt(); 

// Wait until it's clear to go ahead... 
blocker.WaitOne();