Created
July 17, 2015 17:18
-
-
Save joproulx/3a347e7081f589c5301e to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
public static class ObservableExtensions | |
{ | |
public static IObservable<TOut> Select<TIn, TOut>(this IObservable<TIn> source, Func<TIn, Task<TOut>> asyncSelector) | |
{ | |
return Observable.Create<TOut>(observer => | |
{ | |
Task task = Task.FromResult(default(object)); | |
return source.Subscribe(value => | |
{ | |
var asyncProjection = asyncSelector(value); | |
task = Task.WhenAll(task, asyncProjection) | |
.ContinueWith(_ => | |
{ | |
try | |
{ | |
observer.OnNext(asyncProjection.Result); | |
} | |
catch (Exception ex) | |
{ | |
observer.OnError(ex); | |
} | |
}, TaskContinuationOptions.ExecuteSynchronously); | |
}, | |
error => task.ContinueWith(_ => observer.OnError(error), TaskContinuationOptions.ExecuteSynchronously), | |
() => task.ContinueWith(_ => observer.OnCompleted(), TaskContinuationOptions.ExecuteSynchronously)); | |
}); | |
} | |
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Filtering an observable sequence based on an async predicate
Extension methods for Reactive Extension (Rx, C#)
Description
One of the many
Observable.SelectMany()
overloads has the following signature:This operator basically lets you generate an observable sequence by applying an async transformation on each element of an input sequence. The catch is that each element in the output sequence is inserted when the async transformation is completed. Meaning that the output sequence order can be different than the input one.
This is an example of the behavior:
produce a result like this:
In some cases, you don't mind about the order, so the behavior is perfectly valid. But what if you want to keep the same order as the input sequence no matter how long each async transformation takes?
Select
This gists includes an extension method that lets you do just that