Ndtyjky

!!warning: Rx newbie!!

We have multiple price feeds. The requirement is to subscribe to all these feeds and only output the latest tick every 1 sec(throttle)

 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

        }



}

The code have multiple issues. If I call AddToCombinedFeed with the same feed multiple times, the streams will get duplicated to start with. Eg. below

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

This brings me to the question. If I want to remove one price stream from the merged stream, how can I do that?

1. Eradicate the need to resubscribe by applying Switch() technique.
   Your _combinedPriceFeed just switches to the next observable that
   will be supplied by _combinePriceFeedChange.


2. Keep a list to manage your multiple feeds. Create the new observable whenever the list changes and provide it via _combinePriceFeedChange.


3. You should get the logic of an corresponding remove method.


4. If you want a more elegant solution and want to get rid of the subject I think this is possible with Dynamic-Data (MIT-License) from RolandPheasant with Nuget. I never used it before. But I would like to hear from you if you came to an even more elegant solution with this maybe.

Code:

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

    }

}