What are Publish and SelectMany doing in this query?

Question

I’ve got new question about one (or two) Reactive methods. In my scenario I needed an observable sequence capable of suppressing other emitted Tasks while the first Task wasn’t completed, and ended up with something like this:

Observable.Interval(TimeSpan.FromMilliseconds(200))
.Select(x => Observable.FromAsync(async () =>
{
    await Task.Delay(1000);
    // Simulating long running task
    Console.WriteLine(x);
}))
.Publish(x => x.FirstAsync().SelectMany(c => c).Repeat())
.Subscribe();

I tried to Google but I really can’t explain few things:

1. First of all, how that works ?
2. What is exactly on that Reactive sequence that blocks the observable from reaching the subscription? What exactly Replay does in that? Isn’t Replay supposed to replay the Task in this case? Or I don’t know. Can anyone explain detailed every step in that Reactive query? What does Publish with that kind of selector. How Replay is playing in that query? And why do I need to call SelectMany on FirstAsync if only one element will be emitted anyway.

Answer 1

1. The .SelectMany(c => c) is an idiomatic way to flatten/merge a nested sequence. You can replace it with .Merge(), and the behavior of the query will be identical.

2. The Publish operator, when used with a Func<IObservable<TSource>, IObservable<TResult>> parameter, subscribes to the query on which it is chained, and then remains subscribed until the sequence produced by the lambda completes. So in your case, by wrapping the inner sequence x.FirstAsync().SelectMany(c => c).Replay() in a Publish, you delay the unsubscription from the chained sequence (the Interval+Select+FromAsync) until the inner sequence completes. The inner sequence never completes, so the chained sequence keeps forever producing one cold IObservable<Unit> subsequence every second. You can observe this happening, by intercepting a Do operator before the Publish:

.Do(x => Console.WriteLine($"New subsequence: {x.GetType().Name}"))

3. The Replay operator is similar to the Publish, with the difference that the Replay has memory of past notifications, whilst the Publish has no memories whatsoever. I guess that your intention was to attach the Repeat instead of the Replay. The Replay without parameter produces a "connectable" observable, that doesn't subscribe automatically to the chained sequence. You have either to Connect it manually, or to attach the RefCount operator to it. In your case you are doing neither, so the resulting sequence never emits anything and never completes. It's a nasty dead-lock situation.