From IObservable<T> to Task

Question

So the case is this. Suppose somewhere I am filling a Collection. Each time an element is added, an IObservable calls OnNext for its subscribers.

Now, there will be a point where the collection will be filled. (I was reading something and I finished reading .. whatever). At that point, OnComplete() is called on the subscribers.

The user, however, won't observe this method. He will rather call an async method that he will await for ... he doesn't care much about the things he read, he just cares that he finished reading.

So basically, I want, from an IObservable, a Task that returns when the IObservable calls OnComplete() to its subscribers. I specifically want the user not to use an observer, but just to be happy with knowing that whatever happens after his await call will happen after the collection is filled.

Maybe the ToTask() method does the trick? I can't really tell by the documentation.

Just as a side note you should avoid "mixing your monads" - that is changing among IEnumerable<T>, IObservable<T>, and Task<T>. Every change introduces a point in the code that may prematurely force execution and/or introduce unexpected behaviours. Observables are more powerful and expressive than tasks, so it is best to stay in observables. — Enigmativity
Also, it is important to understand that even though observables are called collections they do not get "filled". They exist in the time dimension so they are instances of objects that occur as instances in time. They are a timeline that may or may not end. — Enigmativity
Sounds like you should use Task all the way. You don't have a stream just a single result. — user630190

NeddySpaghetti NeddySpaghetti · Accepted Answer · 2015-02-24T06:05:05

If you are using Rx 2.0 or later you can await IObservable which returns the last item in the observable. I.e. after the observable has completed.

var lastItem = await myObservable;

This is possible because in Rx 2.0 a GetAwaiter extension method on IObservable was added making it possible to await observables. There are also some handy extension methods that allow you to specify which element you want to await.

There is a nice blog about it here.

From IObservable<T> to Task

1 Answers