使用 RX 同步多个事件

Question

假设我在一个列表中有三个产品。为了启用某个动作，所有三个都需要属于某种类型。为了了解产品的类型，我需要拨打服务电话并等待响应。

我想做的是等待所有三个响应（如果出现问题，可能会超时），当收集到所有信息后，决定是否启用可能的操作。

我曾经通过一些计数器或重置事件来跟踪已完成的事件来解决这个问题，但我想看看我是否可以使用 Rx 以更简洁的方式来完成它。

由于我对 Rx 还不太熟悉，所以我正在寻找一些提示/指针。我知道我可以使用

Observable.FromEventPattern

对于我正在等待的事件。我订阅并等待响应并处理它。我只是不清楚如何组合多个事件。

Answer 1

你要找的组合子是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();