IObservable<Task<T>> to IObservable<T> that only produces the latest T

I have an observable stream IObservable<Task<T>> (call it stream A). From this stream, I wish to produce an IObservable<T> (stream B). If these were my only requirements, I believe it would be sufficient to say

streamB = streamA.Select(async x => await x).Select(x => x.Result);

The first Select I expect is truly asynchronous, in the sense that as the stream produces more Task<T> each would be awaited. The second Select however, will then "block" until each of the Task<T>s complete in order. If I'm thinking correctly here, there's no actual blocking going on though, because the second Select won't be entered until each async operation, respectively, in the first Select completes and resumes execution. The problem for me with this pattern is the fact that I must wait for each task to complete (i.e. succeed, fail or be cancelled) in the order that they arrive.

Consider this scenario: Assume I have two Task<T>s, t1 and t2. t1 arrives before t2. However, when t2 arrives, t1 has yet to complete, i.e. in a marble diagram

stream A: -------------------t1-------------t2-------------------------------
<br>
async Select completion --------------------------t1-----------t2------------
<br>
stream B: -----------------------------------------------------t2.Result----

In other words, the arrival of t2 from stream A before the completion of t1 essentially means that t1 should be ignored and not produced by stream B.

I have been able to solve this problem by using very imperative code and adding (too much) complexity, by passing the task to a special helper class that keeps track of all tasks that arrive, using an incrementing long to id each, then awaiting each task and using callbacks that pass the id back to the helper class to tell it which Task has completed. If the id is "older" than the latest to arrive, the result is ignored.

I strongly feel I am over-complicating what on the surface appears as a simple problem. Is there no infrastructure or pattern to solve this kind of problem, either in System.Reactive or e.g. in TPL?

Here's the query you need:

IObservable<T> streamB = streamA.Select(t => Observable.FromAsync(() => t)).Switch();

You should almost always avoid .Select(async x => await x) as it breaks the Rx contract allowing overlapping executions due to the async sending control back to the calling thread.

The normal way to turn an IObservable<Task<T>> to an IObservable<T> is to use .SelectMany(t => Observable.FromAsync(() => t)).

In your case, though, you want to throw away any currently computing values if a new task comes through. So this changes the query from .SelectMany(t => Observable.FromAsync(() => t)) to .Select(t => Observable.FromAsync(() => t)).Switch().

.Switch() turns an IObservable<IObservable<T>> into IObservable<T> by only producing values from the latest inner observable produced by the outer observable. It effectively ignores all but the latest observable. Just what you need.

Here's a demonstration of this working:

void Main()
{
	IObservable<Task<long>> streamA = new []
	{
		ReturnDelayedAsync(1),
		ReturnDelayedAsync(42),
	}.ToObservable();
	
	IObservable<long> streamB = streamA.Select(t => Observable.FromAsync(() => t)).Switch();
	
	streamB.Subscribe(Console.WriteLine);
}

public async Task<long> ReturnDelayedAsync(long x)
{
	await Task.Delay(TimeSpan.FromSeconds(2.0));
	return x;
}

That produces a single value of 42 on the console.