7

我想充分利用.NET (Rx) 的反应式扩展，并希望获得一些关于执行一些基本任务的输入。为了说明我想要做什么，我有一个人为的例子，我有一个带有异步事件的外部组件：

class Component {

  public void BeginStart() { ... }

  public event EventHandler Started;

}

该组件通过调用启动BeginStart()。此方法立即返回，稍后，当组件完成启动时，将触发该Started事件。

我想通过包装组件来创建一个同步启动方法并等待Started事件被触发。到目前为止，这是我想出的：

class ComponentWrapper {

  readonly Component component = new Component();

  void StartComponent() {
    var componentStarted =
      Observable.FromEvent<EventArgs>(this.component, "Started");
    using (var startedEvent = new ManualResetEvent(false))
      using (componentStarted.Take(1).Subscribe(e => { startedEvent.Set(); })) {
        this.componenet.BeginStart();
        startedEvent.WaitOne();
      }
  }

}

我想摆脱ManualResetEvent，并且我希望 Rx 有一个解决方案。但是怎么做？

4

2 回答 2

11

PL 的答案如果非常适合您的规范，但我认为您可能会通过不使用 .First() 与 RX 对抗而获得更好的结果，而是通过为您的组件创建一个 observable 来拥抱它：

    public static IObservable<Unit> AsObservable(this Component component)
    {
        return Observable.Defer(() =>
        {
            component.BeginStart();
            return Observable
                .FromEvent<EventArgs>(component, "Started")
                .Select(_ => new Unit());
        });
    }

然后你可以将它用作阻塞：

new Component().AsObservable().First();

非阻塞：

new Component().AsObservable().Subscribe(_ => Console.WriteLine("Done"));

热的：

var pub = new Component().AsObservable().Publish();
pub.Subscribe(_ => Console.WriteLine("Sub1"));
pub.Subscribe(_ => Console.WriteLine("Sub2"));
pub.Connect();  // started just once per two subscriptions

可组合：

new Component().AsObservable().Delay(TimeSpan.FromSeconds(1));

ETC...

编辑：对于您必须等待并收集信息的多个事件的情况，可以使用以下变体：

public static IObservable<EventArgs> AsObservable(this Component component)
{
    return Observable.Defer(() =>
    {
        component.BeginStart();
        return 
            Observable.FromEvent<EventArgs>(component, "Started1").Take(1)
                .Merge(
            Observable.FromEvent<EventArgs>(component, "Started2").Take(1))
                .Select(evt => evt.EventArgs);
    });
}

有了这个，如果你想阻止直到完成，你可以使用.AsObservable.Last().

于 2010-06-03T06:16:15.260 回答
3

这样的事情应该这样做：

var replay = Observable
    .FromEvent<EventArgs>(this.component, "Started")
    .Replay();
replay.Connect();
component.BeginStart();
replay.First();
于 2010-06-02T19:42:55.250 回答