How to get an exception from a task when a task generates events

Question

I have the following scenario. A Taskthat throws events and can throw an exception:

public event EventHandler<EventArgs> MyEvent;
new Task(() =>
    {
        while (condition)
        {
            // Generate standard .NET event.
            MyEvent(this, new EventArgs());

            // Maybe throw exception.
            if (somethingIsWrong) throw new Exception();
        }
    });

Everything is pretty simple. I listen to events using Observable.FromEvent:

var events =
    Observable.FromEvent<EventArgs>(h => myClass.MyEvent += h,
                                    h => myClass.MyEvent -= h);
events.Subscribe(
    ev => DoSomethingOnNext(ev),
    ex => DoSomethingOnError(ex),
    () => DoSomethingOnCompleted());

All this works fine if no exception is thrown. However, when an exception throws a task, I would like to know this in my observable. The exception is now "hidden" inside the task.

Can I do this by creating another event when an exception occurs, wrapping it in IObservableand subscribing to this new observable? Or is there an easier way?

Answer 1

How about this:

class Program
{
    public event EventHandler<EventArgs> MyEvent;
    static void Main(string[] args)
    {
        var myClass = new Program();            

        var task = new Task(() =>
        {
            for(var i=0; i<5; i++) {
                // Generate standard .NET event. 
                myClass.MyEvent(myClass, new EventArgs());
            }

            throw new Exception();
        });

        var obsTask = task.ToObservable();

        var events = Observable.FromEvent<EventArgs>(h => myClass.MyEvent += h, h => myClass.MyEvent -= h);            

        events.TakeUntil(obsTask).Subscribe(
            ev => DoSomethingOnNext(ev), 
            ex => DoSomethingOnError(ex),
            () => DoSomethingOnCompleted());

        task.Start();

        Console.ReadKey();
    }

    private static void DoSomethingOnCompleted()
    {
        Console.WriteLine("DoSomethingOnCompleted");
    }

    private static void DoSomethingOnError(Exception ex)
    {
        Console.WriteLine("DoSomethingOnError:" + ex.ToString());
    }

    private static void DoSomethingOnNext(IEvent<EventArgs> ev)
    {
        Console.WriteLine("DoSomethingOnNext:" + ev.ToString());
    }

Conclusion:

DoSomethingOnNext: System.Collections.Generic.Event 1[System.EventArgs] DoSomethingOnNext:System.Collections.Generic.Event 1 [System.EventArgs] DoSomethingOnNext: System.Collections.Generic.Event 1[System.EventArgs] DoSomethingOnNext:System.Collections.Generic.Event 1 [System.EventArgs] DoSomethingOnNext: System.Collections.Generic.Event`1 [System.EventArgs] DoSomethingOnError: System.AggregateException: . --- > System.Exception: "System.Exception".   RxDisposeTests.Program. < > c_DisplayClass9.b_0() C:\Users\Richard.Hein\Documents\Visual Studio 2010\Projects\RxConsole\RxDisposeTests\Program.cs: 25   System.Threading.Tasks.Task.InnerInvoke()   System.Threading.Tasks.Task.Execute()  --- --- --- > (Inner Exception # 0) System.Exception: "System.Exception".   RxDisposeTests.Program. < > c_DisplayClass9.b_0() C:\Users\Richard.Hein\Documents\Visual Studio 2010\Projects\RxConsole\RxDisposeTests\Program.cs: 25   System.Threading.Tasks.Task.InnerInvoke()   System.Threading.Tasks.Task.Execute() < ---

EDIT:

, TakeUntil , Task - , , ? :

var events = Observable.CreateWithDisposable<IEvent<EventArgs>>(observer =>
{
    var eventObs = Observable.FromEvent<EventArgs>(
        h => myClass.MyEvent += h, h => myClass.MyEvent -= h);
    task.ToObservable().Subscribe(_ => { }, observer.OnError, observer.OnCompleted);
    return eventObs.Subscribe(observer.OnNext, observer.OnError, observer.OnCompleted);
});

events.Subscribe(
    ev => DoSomethingOnNext(ev), 
    ex => DoSomethingOnError(ex),
    () => DoSomethingOnCompleted());