代码之家 › 专栏 › 技术社区 › Martin Liversage

使用Rx同步异步事件

system.reactive .net c#

Martin Liversage · 技术社区 · 16 年前

我想把 Reactive Extensions for .NET (Rx) 很好的使用和想得到一些投入做一些基本的任务。为了说明我试图做什么,我有一个人为的例子,其中有一个带有异步事件的外部组件:

class Component {

  public void BeginStart() { ... }

  public event EventHandler Started;

}

组件是通过调用 BeginStart() . 此方法立即返回,稍后,当组件完成启动时 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有一个解决方案。但是怎么做呢?

2 回复 | 直到 16 年前

Community Mohan Dere 9 年前

PL's answer 如果完全符合您的规范,但我认为您可以通过不使用.First()来对抗RX,而是通过创建组件的可观察对象来接受它,从而获得更好的结果:

    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));

等。。。

编辑:对于需要等待和收集信息的多个事件, 可以使用以下变化:

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()

PL. 16 年前

像这样的事情应该可以做到:

var replay = Observable
    .FromEvent<EventArgs>(this.component, "Started")
    .Replay();
replay.Connect();
component.BeginStart();
replay.First();