Map Fehler auf beobachtbare mit C# ReactiveX

Question

Ich habe eine beobachtbare MyObservable<Object> die CustomExceptions wo

private class CustomException : Exception 

Was will ich die CustomExceptions in Objekte tun umwandeln werfen können und diejenigen, die in einem neuen beobachtbaren emittieren.

Das ist meine Lösung bisher, aber ich fragte mich, ob dies getan werden könnte, ohne die Methoden OnNext, OnCompleted oder OnError des Subjekts direkt aufrufen zu müssen.

var MySubject = new Subject<NewObject>(); 

MyObservable.Catch<Object, CustomException>(
      ex => 
      { 
       NewObject o = new NewObject(ex.Message); 
       MySubject.OnNext(o); 
       return Observable.Empty<Object>(); 
      }); 

IObservable<IList<NewObject>> listObservable = MySubject.ToList(); 

Bearbeiten: Danke ibebbs! Lief wie am Schnürchen!

Answer 1

Sie können mithilfe der Materialize() Funktion ein Thema fangen und Karten Ausnahmen ohne wie hier gezeigt:

var errorObservable = source 
    .Select(projection) 
    .Materialize() 
    .Where(notification => notification.Kind == NotificationKind.OnError) 
    .Select(notification => notification.Exception) 
    .OfType<CustomException>() 
    .Select(exception => new NewObject(exception.Message)); 

Die Materialize Funktion nimmt eine IObservable<T> und ordnet sie einem IObservable<Notification<T>> wo jede Meldung eine Kind von OnNext hat, OnError oder OnComplete. Das obige Observable sucht einfach nach Benachrichtigungen mit einer Kind`` of OnError and with the Exception being an instance of CustomException then projects these exceptions into an IObservable```.

Hier ist ein Unit-Test zeigt diese Arbeit:

[Fact] 
public void ShouldEmitErrorsToObservable() 
{ 
    Subject<int> source = new Subject<int>(); 
    List<int> values = new List<int>(); 
    List<NewObject> errors = new List<NewObject>(); 

    Func<int, int> projection = 
     value => 
     { 
      if (value % 2 == 1) throw new CustomException("Item is odd"); 

      return value; 
     }; 

    Func<CustomException, IObservable<int>> catcher = null; 

    catcher = ex => source.Select(projection).Catch(catcher); 

    var errorObservable = source 
     .Select(projection) 
     .Materialize() 
     .Where(notification => notification.Kind == NotificationKind.OnError) 
     .Select(notification => notification.Exception) 
     .OfType<CustomException>() 
     .Select(exception => new NewObject(exception.Message)); 

    var normalSubscription = source.Select(projection).Catch(catcher).Subscribe(values.Add); 
    var errorSubscription = errorObservable.Subscribe(errors.Add); 

    source.OnNext(0); 
    source.OnNext(1); 
    source.OnNext(2); 

    Assert.Equal(2, values.Count); 
    Assert.Equal(1, errors.Count); 
} 

Doch wie Sie mit den aufgefasst Fangmechanismen oben eingesetzt sehen können, außer in Rx Handhabung kann schwierig sein Recht und noch schwieriger zu bekommen zu tun elegant. Bedenken Sie stattdessen, dass Exceptions should be Exceptional und wenn Sie eine Fehlerklasse erwarten, dass Sie eine benutzerdefinierte Ausnahme für es geschrieben haben, ist der Fehler nicht wirklich außergewöhnlich, sondern Teil eines Prozessablaufs, der diese Fehler behandeln muss.

In diesem Fall würde ich empfehlen, das Observable in eine Klasse zu projizieren, die die "probiere diese Operation aus und notiere das Ergebnis, sei es ein Wert oder eine Ausnahme" und benutze dies weiter entlang der Ausführungskette.

Im folgenden Beispiel verwende ich eine "Fallible" -Klasse, um das Ergebnis oder die Ausnahme einer Operation zu erfassen und dann einen Strom von "fehlerhaften" Instanzen zu abonnieren und die Fehler von den Werten zu trennen. Wie Sie sehen werden, ist der Code sowohl sauberere und bessere Leistung als sowohl die Fehler und die Werte einer einzelnen Abonnement auf die zugrunde liegende Quelle teilen:

internal class Fallible 
{ 
    public static Fallible<TResult> Try<TResult, TException>(Func<TResult> action) where TException : Exception 
    { 
     try 
     { 
      return Success(action()); 
     } 
     catch (TException exception) 
     { 
      return Error<TResult>(exception); 
     } 
    } 

    public static Fallible<T> Success<T>(T value) 
    { 
     return new Fallible<T>(value); 
    } 

    public static Fallible<T> Error<T>(Exception exception) 
    { 
     return new Fallible<T>(exception); 
    } 
} 

internal class Fallible<T> 
{ 
    public Fallible(T value) 
    { 
     Value = value; 
     IsSuccess = true; 
    } 

    public Fallible(Exception exception) 
    { 
     Exception = exception; 
     IsError = true; 
    } 

    public T Value { get; private set; } 
    public Exception Exception { get; private set; } 
    public bool IsSuccess { get; private set; } 
    public bool IsError { get; private set; } 
} 

[Fact] 
public void ShouldMapErrorsToFallible() 
{ 
    Subject<int> source = new Subject<int>(); 
    List<int> values = new List<int>(); 
    List<NewObject> errors = new List<NewObject>(); 

    Func<int, int> projection = 
     value => 
     { 
      if (value % 2 == 1) throw new CustomException("Item is odd"); 

      return value; 
     }; 

    var observable = source 
     .Select(value => Fallible.Try<int, CustomException>(() => projection(value))) 
     .Publish() 
     .RefCount(); 

    var errorSubscription = observable 
     .Where(fallible => fallible.IsError) 
     .Select(fallible => new NewObject(fallible.Exception.Message)) 
     .Subscribe(errors.Add); 

    var normalSubscription = observable 
     .Where(fallible => fallible.IsSuccess) 
     .Select(fallible => fallible.Value) 
     .Subscribe(values.Add); 

    source.OnNext(0); 
    source.OnNext(1); 
    source.OnNext(2); 

    Assert.Equal(2, values.Count); 
    Assert.Equal(1, errors.Count); 
} 

Hoffe, es hilft.