Captura de excepciones que pueden arrojarse desde una suscripción OnNext Acción

Question

Soy un poco nuevo en Rx.NET. ¿Es posible atrapar una excepción que puede ser lanzada por cualquiera de los suscriptores? Tome lo siguiente ...

handler.FooStream.Subscribe(
      _ => throw new Exception("Bar"), 
      _ => { }); 

Actualmente estoy contactando por suscripción con una instancia de lo siguiente. La implementación de la cual solo usa un ManualResetEvent para activar un hilo en espera.

public interface IExceptionCatcher 
{ 
    Action<T> Exec<T>(Action<T> action); 
} 

y usarlo como tal ...

handler.FooStream.Subscribe(
      _exceptionCatcher.Exec<Foo>(_ => throw new Exception("Bar")), //It's disappointing that this generic type can't be inferred 
      _ => { }); 

siento que tiene que haber alguna manera mejor. ¿Están todas las capacidades de manejo de errores en Rx.NET específicamente para tratar con errores observables?

EDITAR: Por solicitud, mi implementación es https://gist.github.com/1409829 (interfaz e implementación separadas en diferentes ensamblajes en código prod). Comentarios son bienvenidos. Esto puede parecer tonto, pero estoy usando Castle Windsor para administrar muchos suscriptores Rx diferentes. Este receptor excepción se ha registrado en el contenedor como este

windsorContainer.Register(Component.For<IExceptionCatcher>().Instance(catcher)); 

A continuación, se utilizaría como este donde observable es instancia de IObservable:

var exceptionCatcher = 
    new ExceptionCatcher(e => 
           { 
            Logger.FatalException(
             "Exception caught, shutting down.", e); 
            // Deal with unmanaged resources here 
           }, false); 


/* 
* Normally the code below exists in some class managed by an IoC container. 
* 'catcher' would be provided by the container. 
*/ 
observable /* do some filtering, selecting, grouping etc */ 
    .SubscribeWithExceptionCatching(processItems, catcher); 

Answer 1

El built-in operadores observables no hace lo que está preguntando por defecto (al igual que los eventos), pero podría hacer un método de extensión que haría esto.

public static IObservable<T> IgnoreObserverExceptions<T, TException>(
           this IObservable<T> source 
           ) where TException : Exception 
{ 
    return Observable.CreateWithDisposable<T>(
     o => source.Subscribe(
      v => { try { o.OnNext(v); } 
        catch (TException) { } 
      }, 
      ex => o.OnError(ex), 
      () => o.OnCompleted() 
      )); 
} 

Entonces cualquier observable podría ser envuelto por este método para obtener el comportamiento que describió.