1

I'm not sure why, but for some reason when using the observable that is created via concat I will always get all values that are pushed from my list (works as intended). Where as with the normal subscribe it seems that some values never make it to those who have subscribed to the observable (only in certain conditions).

These are the two cases that I am using. Could anyone attempt to explain why in certain cases when subscribing to the second version not all values are received? Are they not equivalent? The intent here is to rewind the stream. What are some reasons that could explain why Case 2 fails while Case 1 does not.

Replay here is just a list of the ongoing stream.

Case 1.

let observable = 
        Observable.Create(fun (o:IObserver<'a>) ->
                        let next b =
                            for v in replay do
                                o.OnNext(v.Head)
                            o.OnNext(b)
                            o.OnCompleted()
                        someOtherObs.Subscribe(next, o.OnError, o.OnCompleted))
let toReturn = observable.Concat(someOtherObs).Publish().RefCount()

Case 2.

let toReturn = 
    Observable.Create(fun (o:IObserver<'a>) ->
        for v in replay do
            o.OnNext(v.Head)
        someOtherObs.Subscribe(o)
    ).Publish().RefCount()
2
  • The functions passed to Observable.Create are different in the two cases. Do you also see the same behaviour if you use exactly the same function in both cases? Mar 5, 2014 at 15:23
  • @MarkSeemann see comment below.
    – Dave
    Mar 5, 2014 at 16:40

1 Answer 1

4

Caveat! I don't use F# regularly enough to be 100% comfortable with the syntax, but I think I see what's going on.

That said, both of these cases look odd to me and it greatly depends on how someOtherObs is implemented, and where (in terms of threads) things are running.

Case 1 Analysis

You apply concat to a source stream which appears to work like this:

  • It subscribes to someOtherObs, and in response to the first event (a) it pushes the elements of replay to the observer.
  • Then it sends event (a) to the observer.
  • Then it completes. At this point the stream is finished and no further events are sent.
  • In the event that someOtherObs is empty or just has a single error, this will be propagated to the observer instead.

Now, when this stream completes, someOtherObs is concatenated on to it. What happens now is a little unpreditcable - if someOtherObs is cold, then the first event would be sent a second time, if someOtherObs is hot, then the first event is not resent, but there's a potential race condition around which event of the remainder will go next which depends on how someOtherObs is implemented. You could easily miss events if it's hot.

Case 2 Analysis

You replay all the replay events, and then send all the events of someOtherObs - but again there's a race condition if someOtherObs is hot because you only subscribe after pushing replay, and so might miss some events.

Comments

In either case, it seems messy to me.

This looks like an attempt to do a merge of a state of the world (sotw) and a live stream. In this case, you need to subscribe to the live stream first, and cache any events while you then acquire and push the sotw events. Once sotw is pushed, you push the cached events - being careful to de-dupe events that may been read in the sotw - until you are caught up with live at which point you can just pass live events though.

You can often get away with naive implementations that flush the live cache in an OnNext handler of the live stream subscription, effectively blocking the source while you flush - but you run the risk of applying too much back pressure to the live source if you have a large history and/or a fast moving live stream.

Some considerations for you to think on that will hopefully set you on the right path.

For reference, here is an extremely naïve and simplistic C# implementation I knocked up that compiles in LINQPad with rx-main nuget package. Production ready implementations I have done in the past can get quite complex:

void Main()
{
    // asynchronously produce a list from 1 to 10
    Func<Task<List<int>>> sotw =
        () => Task<List<int>>.Run(() => Enumerable.Range(1, 10).ToList());
    
    // a stream of 5 to 15
    var live = Observable.Range(5, 10);
    
    // outputs 1 to 15
    live.MergeSotwWithLive(sotw).Subscribe(Console.WriteLine);
}

// Define other methods and classes here
public static class ObservableExtensions
{
    public static IObservable<TSource> MergeSotwWithLive<TSource>(
        this IObservable<TSource> live,
        Func<Task<List<TSource>>> sotwFactory)
    {
        return Observable.Create<TSource>(async o =>
        {       
            // Naïve indefinite caching, no error checking anywhere             
            var liveReplay = new ReplaySubject<TSource>();
            live.Subscribe(liveReplay);
            // No error checking, no timeout, no cancellation support
            var sotw = await sotwFactory();
            foreach(var evt in sotw)
            {
                o.OnNext(evt);
            }                               
                        
            // note naive disposal
            // and extremely naive de-duping (it really needs to compare
            // on some unique id)
            // we are only supporting disposal once the sotw is sent            
            return liveReplay.Where(evt => !sotw.Any(s => s.Equals(evt)))
                    .Subscribe(o);                  
        });
    }
}
6
  • Thanks for the explanation. The data is hot but is coming in a controlled manner via a Synchronous Subject (this is the live stream). I don't know why but when I removed Publish().RefCount() from Case 2 I got the intended results.
    – Dave
    Mar 5, 2014 at 16:38
  • 1
    I wouldn't rely on that fix. I've added a C# example that shows a bit more of what's involved. I've tried to indicate where it's being overly simplistic. I think a real-world example would be overly complex for this forum. Maybe I've blog one at some point! Mar 5, 2014 at 17:08
  • shouldn't the live replay be subscribed to the observer?
    – Dave
    Mar 5, 2014 at 17:25
  • Is there anyway to have a replay subject only cache until it has been subscribed to? You would think this was common.
    – Dave
    Mar 5, 2014 at 21:42
  • 1
    No. You can specify a limit on the cache though. In my production implementations I don't use a subject. It was just easy to demo the points here. Mar 6, 2014 at 0:54

Your Answer

Reminder: Answers generated by Artificial Intelligence tools are not allowed on Stack Overflow. Learn more

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.