处理合并的内部订阅

Question

！！警告：Rx 新手！！

我们有多个价格信息。要求是订阅所有这些提要，并且每 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
        }

}

代码有多个问题。如果我多次使用相同的提要调用 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 提供。
   1. 保留一个列表以管理您的多个提要。每当列表更改时创建新的 observable，并通过 _combinePriceFeedChange 提供。
   2. 你应该得到对应的remove方法的逻辑。

代码：

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);
}