c# - 如何使用 Reactive Extensions 实现事件

Question

Reactive Extensions 允许您使用 Observable.FromEventPattern 轻松订阅事件, 但我找不到任何关于当你有一个 IObservable 时你如何实现一个事件的信息.

我的情况是这样的:我需要实现一个包含事件的接口(interface)。每当我的对象的某个值发生变化时，应该调用该事件，并且出于线程安全原因，我需要在某个 SynchronizationContext 上调用此事件。 .我还应该在注册时使用当前值调用每个事件处理程序。

public interface IFooWatcher
{
    event FooChangedHandler FooChanged;
}

使用 Rx 使用 BehaviorSubject 获得一个符合我要求的可观察对象相当容易:

public class FooWatcher
{
    private readonly BehaviorSubject<Foo> m_subject;
    private readonly IObservable<Foo> m_observable;

    public FooWatcher(SynchronizationContext synchronizationContext, Foo initialValue)
    {
        m_subject = new BehaviorSubject<Foo>(initialValue);
        m_observable = m_subject
            .DistinctUntilChanged()
            .ObserveOn(synchronizationContext);
    }

    public event FooChangedHandler FooChanged
    {
        add { /* ??? */ }
        remove { /* ??? */ }
    }
}

现在我正在寻找一种简单的方法来获得 add和 remove函数订阅和取消订阅传递的 FooChangedHandler作为Observer<Foo>在 m_observable .我当前的实现与此类似:

    add
    {
        lock (m_lock)
        {
            IDisposable disp = m_observable.Subscribe(value);
            m_registeredObservers.Add(
                new KeyValuePair<FooChangedHandler, IDisposable>(
                    value, disp));
        }
    }

    remove
    {
        lock (m_lock)
        {
            KeyValuePair<FooChangedHandler, IDisposable> observerDisposable =
                m_registeredObservers
                    .First(pair => object.Equals(pair.Key, value));
            m_registeredObservers.Remove(observerDisposable);
            observerDisposable.Value.Dispose();
        }
    }

但是，我希望找到一个更简单的解决方案，因为我需要实现其中几个事件(不同处理程序类型)。我尝试推出自己的通用解决方案，但它会产生一些需要解决的额外问题(特别是，您通常如何使用采用参数 T 的委托(delegate))，所以我更愿意找到现有的解决方案弥合了这个方向的差距 - 正如FromEventPattern做相反的事情。

Answer 1

你可以这样做:

public event FooChangedHandler FooChanged
{
    add { m_observable.ToEvent().OnNext += value; }
    remove { m_observable.ToEvent().OnNext -= value; }
}

但是，在删除时，我认为您可能只是想处理订阅...或者从 ToEvent() 获取 Action 并将其存储为成员。未经测试。

编辑:但是，您必须使用 Action 而不是 FooChangedHandler 委托(delegate)。

编辑 2:这是一个经过测试的版本。但是，我想您需要使用 FooChangedHandler，因为您有一堆这些预先存在的处理程序？

void Main()
{
    IObservable<Foo> foos = new [] { new Foo { X = 1 }, new Foo { X = 2 } }.ToObservable();
    var watcher = new FooWatcher(SynchronizationContext.Current, new Foo { X = 12 });
    watcher.FooChanged += o => o.X.Dump();  
    foos.Subscribe(watcher.Subject.OnNext); 
}

// Define other methods and classes here

//public delegate void FooChangedHandler(Foo foo);
public interface IFooWatcher
{
    event Action<Foo> FooChanged;
}

public class Foo {
    public int X { get; set; }
}
public class FooWatcher
{

    private readonly BehaviorSubject<Foo> m_subject;
    public BehaviorSubject<Foo> Subject { get { return m_subject; } }
    private readonly IObservable<Foo> m_observable;

    public FooWatcher(SynchronizationContext synchronizationContext, Foo initialValue)
    {
        m_subject = new BehaviorSubject<Foo>(initialValue);

        m_observable = m_subject
            .DistinctUntilChanged();
    }

    public event Action<Foo> FooChanged
    {
        add { m_observable.ToEvent().OnNext += value; }
        remove { m_observable.ToEvent().OnNext -= value; }
    }
}