1

所以我正在尝试使用 Rx.NET 组合一个小消息总线。

public class Bus {
    private readonly Subject<BaseCommand> _commands = new Subject<BaseComand>();

    public void RegisterHandler<TCommand>(Action<TCommand> handler) where TCommand: BaseCommand {
        _commands
            .OfType&lt;TCommand&gt;()
            .Publish()
            .RefCount()
            .Subscribe(handler);
    }

    public void SendCommand<TCommand>(TCommand command) where TCommand: BaseCommand {
        _commands.OnNext(command);
    }
}

这就是代码的要点。我想限制订阅，以便单个消息类型只能存在一个订阅。OfType<T>在添加新订阅之前，是否有任何现有订阅检查 Observable 的方法？

4

1 回答 1

1

我会建议这样的事情（我已将您的 RegisterHandler 更改为具有 IDisposable 返回类型，因此您实际上可以再次取消订阅）：

public class Bus
{
    private readonly Subject<BaseCommand> _commands = new Subject<BaseCommand>();

    private class Counter<TCommand> where TCommand : BaseCommand
    {
        public static int Count;
    }

    public IDisposable RegisterHandler<TCommand>(Action<TCommand> handler, Action<Exception> OnError = null) where TCommand : BaseCommand
    {
        OnError = OnError ?? (Action<Exception>)((ex) => Dispatcher.CurrentDispatcher.Invoke(() => {throw ex; })); // alternative case of course only works if dispatcher is available

        return Observable.Create<TCommand>(o =>
        {
            if (Interlocked.Increment(ref Counter<TCommand>.Count) > 1)
            {
                Interlocked.Decrement(ref Counter<TCommand>.Count);
                o.OnError(new InvalidOperationException("Too many subscribers!"));
                return Disposable.Empty;
            }

            var subscription = _commands
                .OfType<TCommand>()
                      .Publish()
                      .RefCount()
                      .Subscribe(o);

            var decrement = Disposable.Create(() =>
            {
                Interlocked.Decrement(ref Counter<TCommand>.Count);
            });

            return new CompositeDisposable(subscription, decrement);
        })
        .Subscribe(handler, OnError);

    }

    public void SendCommand<TCommand>(TCommand command) where TCommand : BaseCommand
    {
        _commands.OnNext(command);
    }
}

编辑：我可能会将您的 RegisterHandler 函数的签名更改为

 public IObservable<TCommand> RegisterHandler<TCommand>() where TCommand : BaseCommand

尽管; 节省了一些错误管理的麻烦（订阅者必须自己处理），并且您的消费者在订阅这些事件的时间和方式上更加自由。

于 2017-04-27T19:33:00.667 回答