使用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();