带有Rx的EventBus

Question

您如何使用Rx编写类似于以下简单EventBus示例的内容？

public class EventBus
{
    private readonly Dictionary<Type, List<Action<Event>>> routes = new Dictionary<Type, List<Action<Event>>>();

    public void RegisterHandler<T>(Action<T> handler) where T : Event
    {
        List<Action<Event>> handlers;

        if (!this.routes.TryGetValue(typeof(T), out handlers))
        {
            handlers = new List<Action<Event>>();
            this.routes.Add(typeof(T), handlers);
        }

        handlers.Add(x => handler((T)x));
    }

    public void Publish<T>(T @event) where T : Event
    {
        List<Action<Event>> handlers;

        if (!this.routes.TryGetValue(@event.GetType(), out handlers))
        {
            return;
        }

        foreach (var handler in handlers)
        {
            var apply = handler;
            apply(@event);
        }
    }
}

Answer 1

由于你的功能EventBus是相当基本的,我认为这样的东西可能对你来说足够好:

public class EventBus
{
    private readonly Subject<Event> subject = new Subject<Event>();

    public IObservable<T> GetEventStream<T>() where T : Event
    {
        return subject.OfType<T>();
    }

    public void Publish<T>(T @event) where T : Event
    {
        subject.OnNext(@event);
    }
}

在该解决方案中Action<T>,您只需订阅事件流,而不是注册:

var bus = new EventBus();
var disposable = bus.GetEventStream<SomeEvent>.Subscribe(ev => Console.WriteLine("It happened!"));
// some time later
bus.Publish(new SomeEvent());
// you can also unsubscribe from the stream
disposable.Dispose();