Création d'une méthode d'extension observable non de blocage qui renvoie un élément par défaut pour une séquence vide

Question

Imaginez la LINQ suivante à la déclaration observables:

var x = from result1 in service1.operation() 
     from result2 in service2.operation() 
     from result3 in service3.operation() 
     select DoSomething() 

x.Subscribe() 

void Unit DoSomething() { 
... 
} 

Les services renvoient aussi un observable froid, donc ils attendront chacun à compléter, puis DoSomething est appelé.

Maintenant, service2.operation renvoie un Observable.Empty qui est essentiellement une notification incomplète, ce qui signifie que service3 ne sera jamais appelé et DoSomething non plus. Je voulais qu'il continue la chaîne si un oncomplete est retourné, mais fournissez une valeur par défaut pour result2. Donc, j'ai créé une méthode d'extension OnEmptyReturnDefault

public static IObservable<T> OnEmptyReturnDefault<T>(this IObservable<T> observable) 
{ 
    var maybeReturnsSomething = observable.Memoize(); // Custom Lazy caching extension method 
    var whenEmpty = from isEmpty in maybeReturnsSomething.IsEmpty() 
        where isEmpty 
        select default(T); 
    var whenNotEmpty = from isEmpty in maybeReturnsSomething.IsEmpty() 
         where !isEmpty 
         from notEmpty in maybeReturnsSomething 
         select notEmpty; 
    return whenEmpty.Merge(whenNotEmpty); 
} 

Permettant moi de le faire:

var x = from result1 in service1.operation() 
      from result2 in service2.operation().OnEmptyReturnDefault() 
      from result3 in service3.operation() 
      select DoSomething() 

Tout est bon, sauf que ma solution bloque. IsEmpty() fait essentiellement Take (1) .Count() == 0. Je veux une solution qui ne bloque pas.

Answer 1

Une solution proposée sur un autre site fonctionne bien:

public static IObservable<T> DefaultIfEmpty<T>(this IObservable<T> src, T defaultValue) 
{ 
    return src 
    .Materialize() 
    .Select((n, i) => (n.Kind == NotificationKind.OnCompleted && i == 0) 
       ? new Notification<T>[] 
       { 
        new Notification<T>.OnNext(defaultValue), 
        new Notification<T>.OnCompleted() 
       } 
       : new[] {n}) 
    .SelectMany(ns => ns) 
    .Dematerialize(); 
}