DoAsync with Rx

Question

First, is there an implementation like "DoAsync" to use with Rx? Considering that I have a specific SynchronizationContext and IScheduler using the same thread, some like https://gist.github.com/OmerMor/1554548

Now, see the code below:

  public static IObservable<T> DoAsyncWithFallBack<T>(this IObservable<T> source, Func<T, Task> accessor, Action<T, Exception> localizedFallback)
    {
        return new AnonymousObservable<T>(obs =>
        {
            return source.Subscribe(async x =>
            {
                try
                {
                    await accessor(x);
                    obs.OnNext(x);
                }
                catch (Exception ex)
                {
                    localizedFallback(x, ex);
                    obs.OnError(ex);
                }
            }, obs.OnError, obs.OnCompleted);
        });
    } 

and the usage (just for example):

   Observable.Return(1)
             .ObserveOn(scheduler)
             .Select(a => new
                             {
                                 EventData = a,
                                 Task = TaskEx.Run(() => DoSomething(a))
                             })
             .DoAsyncWithFallBack(async a => await a.Task, (a, ex) => FallBackPlan(a.EventData, ex))
             .Subscribe(next => {}, ex => {});

Using this code above, if I get an exception, the stack trace does not help me at all, I lose the information about the DoSomething method, what would be the right way to get it?

If you search a little you can find some problems about exceptions with async/await: Is it possible to get a good stack trace with .NET async methods?

Answer 1

For .NET 4.0, your best bet is to wrap the exception before passing it to OnError. This will preserve the stack trace of the exception.

obs.OnError(new ApplicationException("Error in DoAsync", ex));