Cómo manejar excepciones de métodos asíncronos dentro de una instrucción SelectMany

Question

Estoy tratando de procesar algunas tareas de forma asincrónica con Rx, p. Ej.

var list = Enumerable.Range(0, 100) 
    .ToObservable() 
    .SelectMany(x => Observable.Start(() => { 
     Console.WriteLine("Processing {0} ...", x); 

     Thread.Sleep(100 * x % 3); 

     if (x > 90) { 
      Console.WriteLine("Procesing exception {0} > 90", x); 
      throw new Exception("Value too large"); 
     } 
     Console.WriteLine("Processing {0} completed.", x); 
     return x; 
    })) 
    .Subscribe(
     x => { Console.WriteLine("Next [{0}]", x); }, 
     e => { 
      Console.WriteLine("Exception:"); 
      Console.WriteLine(e.Message); 
     }, 
     () => { Console.WriteLine("Complete"); } 
    ); 

El problema que tengo con este código es que la excepción no se pasa al suscriptor. Entonces, después de intentarlo, me rendí y decidí hacer esta simple pregunta:

¿Cómo se manejan las excepciones planteadas desde los métodos asíncronos dentro de una declaración SelectMany?

Solo para dejarlo en claro, la implementación final es una llamada de función sincrónica que puede lanzar o no una excepción. El objetivo es pasarlo al suscriptor para que pueda seguir procesándolo (en el caso específico, un mensaje se mostrará al usuario).

Editar

Moví los hallazgos a una respuesta, de modo que pueda marcar esta pregunta como respondida. Personalmente, no estoy de acuerdo con mi propia respuesta ... pero a veces no hay otra manera, lo siento mucho.

Answer 1

La respuesta

En realidad el código es de funcionar correctamente. Sin embargo, el depurador se rompe en las excepciones ya que las operaciones asincrónicas se siguen ejecutando en segundo plano, al menos las que ya se iniciaron cuando se produjo la primera excepción. ¡Me arrojó! Si ejecuta el código sin depurador las excepciones son tragadas. Así que supongo que el problema estaba realmente en frente de la computadora :-)

Todavía tengo algunas aclaraciones sobre el Observable.Start, como suponía, y esto correctamente, que la implementación debería tener en realidad, se implementó algún tipo de error ... ver Antecedentes.

Antecedentes

Observable.Start es un método de conveniencia que utiliza el método Observable.ToAsync para activar una función/acion en una operación asíncrona. Si observa la implementación del método, verá que ya hace el manejo/reenvío de excepciones.

public static Func<IObservable<TResult>> ToAsync<TResult>(this Func<TResult> function, IScheduler scheduler) { 
    if (function != null) { 
     if (scheduler != null) { 
      return() => { 
       AsyncSubject<TResult> asyncSubject = new AsyncSubject<TResult>(); 
       scheduler.Schedule(() => { 
        TResult result = default(TResult); 
        try { 
         result = function(); 
        } catch (Exception exception1) { 
         Exception exception = exception1; 
         asyncSubject.OnError(exception); 
         return; 
        } 
        asyncSubject.OnNext(result); 
        asyncSubject.OnCompleted(); 
       }); 
       return asyncSubject.AsObservable<TResult>(); 
      }; 
     } else { 
      throw new ArgumentNullException("scheduler"); 
     } 
    } else { 
     throw new ArgumentNullException("function"); 
    } 
} 

Answer 2

Utilice Materialize para convertir sus mensajes OnError/OnCompleted en notificaciones.

Por ejemplo,

observable.SelectMany(x => Observable.Start(fn).Materialize())

le conseguirá el error/la ejecución envuelto en una notificación para ser manipulados en su camino a punto de las suscripciones efectuadas por la línea, en comparación con el error de ser terminado dentro de la SelectMany .

Esto es útil para la mayoría de las operaciones de llamada Async porque el método falla o se completa.