Vfrdtyky

I need to have a hot observable which wraps a price feed. This is subscribed in multiple areas.

The observable is created using Refcount and passed around for subscription.The first subscriber calls subscribe hence starts the stream and get all the events. The second will miss the events until it's subscription and subsequent subscriptions will do the same behaviour.

It is not the missing of events which is my issue. I want all subscribers to get the same data. That is, the stream must start only when the subscription requests are finished.

Is it possible?

Edit: Two approaches and its problems are illustrated below.

1)

public void HotObservableSubscriptionWithRefCount()

{       

    var obs1 = Observable.Interval(TimeSpan.FromMilliseconds(500)).Select(x =>

                {

                    var publishVal = x;

                    Console.WriteLine($@"observer1 publishing {publishVal}");

                    return publishVal;

                }).Publish().RefCount();



    var obs2 = Observable.Interval(TimeSpan.FromMilliseconds(500)).Select(x =>

                {

                    var publishVal = x + 100;

                    Console.WriteLine($@"observer2 publishing {publishVal}");

                    return publishVal;

                }).Publish().RefCount();



   var sub1 = obs1.Subscribe(x => Console.WriteLine($@"subscriber1 value {x}"));

                var sub2 = obs2.Subscribe(x => Console.WriteLine($@"subscriber2 

             value {x}"));



    Thread.Sleep(TimeSpan.FromSeconds(1));



var combinedSub = obs1.Merge(obs2).Subscribe(x => Console.WriteLine($@"combined 

              subscriber value {x}"));



  Thread.Sleep(TimeSpan.FromSeconds(1));



   sub1.Dispose();

   sub2.Dispose();

   combinedSub.Dispose();



   Thread.Sleep(TimeSpan.FromSeconds(1));

}

Problem: The combined subscriber is missing values from two observables because of the delay in subscription

2)

 public void HotObservableSubscriptionWithPublish()

{

    var obs1 = Observable.Interval(TimeSpan.FromMilliseconds(500)).Select(x =>

                {

                    var publishVal = x;

                    Console.WriteLine($@"observer1 publishing {publishVal}");

                    return publishVal;



                }).Publish();



    var obs2 = Observable.Interval(TimeSpan.FromMilliseconds(500)).Select(x =>

                {

                    var publishVal = x + 100;

                    Console.WriteLine($@"observer2 publishing {publishVal}");

                    return publishVal;

                }).Publish();



    var sub1 = obs1.Subscribe(x => Console.WriteLine($@"subscriber1 value 

               {x}"));

    var sub2 = obs2.Subscribe(x => Console.WriteLine($@"subscriber2 value 

               {x}"));



             Thread.Sleep(TimeSpan.FromSeconds(1));



    var combinedSub = obs1.Merge(obs2).Subscribe(x => 

                Console.WriteLine($@"combined subscriber value {x}"));



    obs1.Connect();

    obs2.Connect();



   Thread.Sleep(TimeSpan.FromSeconds(1));



   sub1.Dispose();

   sub2.Dispose();

   combinedSub.Dispose();



   Thread.Sleep(TimeSpan.FromSeconds(1));

}

This will make sure combinedsubsciber will get values in line with any individual subscribers. However even the subscribers are disposed of, the observer still continue provide values.

I need full life cycle control of publisher and subscriber