Использование Rx для синхронизации асинхронных событий - PullRequest
Новая
       10

Использование Rx для синхронизации асинхронных событий

7 голосов
/ 02 июня 2010

Я хочу использовать Reactive Extensions для .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 есть решение. Но как?

Ответы [ 2 ]

10 голосов
/ 03 июня 2010

Ответ PL , если он идеально подходит для вашей спецификации, но я подумал, что вы могли бы добиться лучших результатов, не сражаясь с RX с .First (), а используя его, создавая наблюдаемый для вашего компонента: 

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

3 голосов
/ 02 июня 2010

Что-то вроде этого должно сделать это: 

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