Слияние меняющейся коллекции наблюдаемых

Question

У нас есть класс Thing, который реализует IObservable<Thing>. В другом классе есть коллекция Thing, и этот класс должен реагировать на обновления из всех этих наблюдаемых единым образом. Очевидный способ сделать это - Observable.Merge(), и это вообще работает; однако, когда коллекция меняется, мы также должны подписаться на любые новые Thing s в нашей объединенной подписке (и, теоретически, отказаться от подписки на все удаленные, но это кажется менее проблематичным - они больше не будут производить никаких обновлений) ,

В настоящее время мы достигаем этого путем воссоздания подписки на каждое изменение коллекции, но это кажется довольно субоптимальным с точки зрения затрат на обработку, а также из-за отсутствия обновлений от любого из Thing в течение короткого времени между отбрасыванием старой подписки и создание нового (что доказало свою эффективность на практике, особенно, поскольку нам также необходимо дозвониться до подписки на короткий промежуток времени, а буферизованные элементы теряются при утилизацию подписки).

Каков надлежащий способ слияния изменяющейся коллекции наблюдаемых?

Answer 1

Если у вас есть IObservable<IObservable<T>> observable, а затем позвоните по номеру Merge, в который войдут дети новых родителей, если вы поймаете мой дрейф. Трюк превращает ObservableCollection<IObservable<Thing>> в IObservable<IObservable<Thing>>.

Если у вас есть ReactiveUI, и вы можете его использовать, вы можете конвертировать ObservableCollection<IObservable<Thing>> в ReactiveCollection<IObservable<Thing>>. ReactiveCollection унаследовал от ObservableCollection, а также осуществляет IObservable.

Если ReactiveUI находится вне вопроса (который я предполагаю, что это происходит потому, что вы уже используете коллекцию Caliburn Micro), то вы можете конвертировать с помощью событий ObservableCollection «s:

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>>()); 

Вот некоторые примеры кода демонстрируют использование:

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 }); 

Используя следующий класс:

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