配置内部订阅合并

Question

!!警告：Rx新手！

我们有多个价格Feed。要求是订阅所有这些订阅源，并且每1秒仅输出一次最新的订单（节流）

 public static class FeedHandler
{
        private static IObservable<PriceTick> _combinedPriceFeed = null;

          private static double _throttleFrequency = 1000;

        public static void AddToCombinedFeed(IObservable<PriceTick> feed)
        {
            _combinedPriceFeed = _combinedPriceFeed != null ? _combinedPriceFeed.Merge(feed) : feed;
            AddFeed(_combinedPriceFeed);
        }

              private static IDisposable _subscriber;

        private static void AddFeed(IObservable<PriceTick> feed)
        {
            _subscriber?.Dispose();
            _subscriber = feed.Buffer(TimeSpan.FromMilliseconds(_throttleFrequency)).Subscribe(buffer => buffer.GroupBy(x => x.InstrumentId, (key, result) => result.First()).ToObservable().Subscribe(NotifyClient));
        }

         public static void NotifyClient(PriceTick tick)
        {
        //Do some action
        }

}

代码有多个问题。如果我多次使用相同的Feed调用AddToCombinedFeed，则流将重复开始。例如。下面

IObservable<PriceTick> feed1;

FeedHandler.AddToCombinedFeed(feed1);//1 stream
FeedHandler.AddToCombinedFeed(feed1);//2 streams(even though the groupby and first() functions will prevent this effect to propagate further

这让我想到了这个问题。如果我想从合并流中删除一个价格流，我该怎么做？

Answer 1

更新 - 新解决方案

来自RolandPheasant和Nuget的Dynamic-Data（MIT-License）。

1. 使用SourceList而不是List
2. 使用MergeMany运算符

码：

public class FeedHandler
{
    private readonly IDisposable _subscriber;
    private readonly SourceList<IObservable<PriceTick>> _feeds = new SourceList<IObservable<PriceTick>>();
    private readonly double _throttleFrequency = 1000;

    public FeedHandler()
    {
        var combinedPriceFeed = _feeds.Connect().MergeMany(x => x).Buffer(TimeSpan.FromMilliseconds(_throttleFrequency)).SelectMany(buffer => buffer.GroupBy(x => x.InstrumentId, (key, result) => result.First()));
        _subscriber = combinedPriceFeed.Subscribe(NotifyClient);
    }

    public void AddFeed(IObservable<PriceTick> feed) => _feeds.Add(feed);

    public void NotifyClient(PriceTick tick)
    {
        //Do some action
    }
}

旧解决方案

1. 通过应用Switch（）技术消除重新订阅的需要。您的_combinedPriceFeed只是切换到_combinePriceFeedChange将提供的下一个observable。 保留一个列表来管理您的多个Feed。每当列表更改时创建新的observable并通过_combinePriceFeedChange提供它。 您应该获得相应的删除方法的逻辑。

码：

public class FeedHandler
{
    private readonly IDisposable _subscriber;
    private readonly IObservable<PriceTick> _combinedPriceFeed;
    private readonly List<IObservable<PriceTick>> _feeds = new List<IObservable<PriceTick>>();
    private readonly BehaviorSubject<IObservable<PriceTick>> _combinedPriceFeedChange = new BehaviorSubject<IObservable<PriceTick>>(Observable.Never<PriceTick>());
    private readonly double _throttleFrequency = 1000;

    public FeedHandler()
    {
        _combinedPriceFeed = _combinedPriceFeedChange.Switch().Buffer(TimeSpan.FromMilliseconds(_throttleFrequency)).SelectMany(buffer => buffer.GroupBy(x => x.InstrumentId, (key, result) => result.First()));
        _subscriber = _combinedPriceFeed.Subscribe(NotifyClient);
    }

    public void AddFeed(IObservable<PriceTick> feed)
    {
        _feeds.Add(feed);
        _combinedPriceFeedChange.OnNext(_feeds.Merge());
    }


    public void NotifyClient(PriceTick tick)
    {
        //Do some action
    }
}

Answer 2

这是您需要的代码：

private static SerialDisposable _subscriber = new SerialDisposable();

private static void AddFeed(IObservable<PriceTick> feed)
{
    _subscriber.Disposable =
        feed
            .Buffer(TimeSpan.FromMilliseconds(_throttleFrequency))
            .SelectMany(buffer =>
                buffer
                    .GroupBy(x => x.InstrumentId, (key, result) => result.First()))
            .Subscribe(NotifyClient);
}