c# - 通过多个订阅者的 'routing' 事件模式？

Question

我有一组具有以下接口(interface)的类:

public interface RoutedEventReceiver<T>
{
    IDisposable Apply(IObservable<T> stream);
    bool ShouldForwardEvent(T anEvent);
}

我想做的是维护这些类的堆栈，每个事件都通过 ShouldForwardEvent(T) 过滤谓词，以及结果 IObservable<T>传递给下一个接收者。我还希望能够在我的程序运行时推送和弹出新的接收器(在某些时候我可能想从一个堆栈移动到其他一些集合，但现在一个堆栈就足够了)。

我目前拥有的确实有效，但我不觉得它很“Rx”。我确信必须有一种方法可以在没有所有这些命令式逻辑的情况下做我想做的事:

private void Refresh()
{
    // _subscriptions is a list of previous subscriptions
    foreach (var subscription in _subscriptions)
        subscription.Dispose();
    _subscriptions.Clear();

    // _stream is my stream of incoming events
    if (_stream != null)
    {
        var stream = _stream;

        foreach (var eventReceiver in _eventReceivers)
        {
            // add the subscription so it can be disposed next Refresh()
            _subscriptions.Add(eventReceiver.Apply(stream));

            // filter the stream for the next event receiver
            stream = stream.Where(eventReceiver.ShouldForwardEvent);
        }
    }
}

每当我 Push 时都会调用上述方法或 Pop在堆栈上。

是否有更简洁、更实用的方式来表达上述意图？我试过.Publish()但收效甚微 - 也许我还不够了解。

Answer 1

我已经设法使 Publish 方法起作用，但除了摆脱保留 IDisposables 列表的需要外，它并没有给我太多帮助:

private void Refresh()
{
    _published.DisposeIfNotNull();

    if (_stream != null)
    {
        var connectable = _stream.Publish();
        _published = connectable.Connect();
        var stream = connectable.AsObservable();

        foreach (var eventReceiver in _eventReceivers)
        {
            eventReceiver.Apply(stream);
            stream = stream.Where(eventReceiver.ShouldForwardEvent);
        }
    }
}