Eine sich ändernde Sammlung von Observablen zusammenführen

Question

Wir haben eine Klasse Thing, die IObservable<Thing> implementiert. In einer anderen Klasse gibt es eine Sammlung von Thing s, und diese Klasse muss auf Aktualisierungen von all diesen Observablen in einheitlicher Weise reagieren. Der offensichtliche Weg, dies zu tun, ist Observable.Merge(), und das funktioniert im Allgemeinen; aber, wenn die Sammlung ändert, müssen wir auch alle neuen Thing s in unserem fusionierten Abonnement abonnieren (und theoretisch von allen entfernten abbestellen, aber das scheint weniger problematisch - sie werden nur keine Updates mehr produzieren) . Das erreichen wir derzeit, indem wir das Abonnement bei jeder Änderung der Sammlung neu erstellen, aber das scheint in Bezug auf den Verarbeitungsaufwand und wegen fehlender Aktualisierungen von irgendeinem der Thing s in der kurzen Zeit zwischen dem Verwerfen des alten Abonnements suboptimal zu sein und das Erstellen des neuen (was sich in der Praxis als ein Problem herausgestellt hat, insbesondere, da wir auch das Abonnement für eine kurze Zeitspanne benötigen und die gepufferten Elemente verloren gehen, wenn das Abonnement entsorgt wird).

Was ist der richtige Weg, eine sich verändernde Sammlung von Observablen wie diese zusammenzuführen?

Answer 1

Wenn Sie eine IObservable<IObservable<T>> observable haben, dann ruft Merge auf, Kinder von neuen Eltern, wenn Sie meine Drift fangen. Der Trick besteht darin, die ObservableCollection<IObservable<Thing>> in eine IObservable<IObservable<Thing>> umzuwandeln.

Wenn Sie ReactiveUI ausgeführt haben und in Ordnung sind, können Sie die ObservableCollection<IObservable<Thing>> in eine ReactiveCollection<IObservable<Thing>> konvertieren. ReactiveCollection erbt von ObservableCollection und implementiert auch IObservable.

Wenn ReactiveUI indiskutabel ist (was ich vermute, es liegt daran, dass Sie bereits eine Caliburn Micro Sammlung verwenden), dann können Sie mit ObservableCollection ‚s Ereignisse konvertieren:

ObservableCollection<IObservable<Thing>> observableCollection = new ObservableCollection<IObservable<Thing>>(); 
IObservable<IObservable<Thing>> oCollectionObservable = Observable.FromEventPattern<NotifyCollectionChangedEventHandler, NotifyCollectionChangedEventArgs>(
    h => observableCollection.CollectionChanged += h, 
    h => observableCollection.CollectionChanged -= h 
) 
    .SelectMany(ep => ep.EventArgs.NewItems.Cast<IObservable<Thing>>()); 

hier ist ein Beispielcode demonstriert Verwendung:

oCollectionObservable 
    .Merge() 
    .Subscribe(t => Console.WriteLine($"Received Thing {{Id = {t.Id}}}")); 

var firstObservable = Observable.Range(1, 5) 
    .Select(i => new Thing { Id = i }) 
    .Concat(
     Observable.Range(8, 5) 
      .Select(i => new Thing { Id = i }) 
      .Delay(TimeSpan.FromSeconds(2)) 
    ); 

observableCollection.Add(firstObservable); 
var subject = new Subject<Thing>(); 
observableCollection.Add(subject); 
subject.OnNext(new Thing { Id = 6 }); 
subject.OnNext(new Thing { Id = 7 }); 

die folgende Klasse verwenden:

public class Thing 
{ 
    public int Id { get; set; } 
}