合并不断变化的可观察量集合

我们有一堂课Thing实现IObservable<Thing>。在另一个类中，有一个集合Things ，并且该类需要以统一的方式对所有这些可观察量的更新做出反应。最明显的方法是Observable.Merge()，这通常有效；however，当集合发生变化时，我们还需要订阅任何新的Things 在我们的合并订阅中（理论上取消订阅所有已删除的订阅，但这似乎问题不大 - 它们只是不再产生任何更新）。

目前，我们通过在集合的每次更改时重新创建订阅来实现这一点，但这在处理开销方面似乎不是最理想的，而且还由于缺少来自任何一个的更新。Thing在放弃旧订阅和创建新订阅之间的短暂时间内（这在实践中已被证明是一个问题，特别是因为我们还需要Buffer()订阅时间很短，并且在处理订阅时缓冲的项目会丢失）。

合并这样不断变化的可观察量集合的正确方法是什么？

如果你有一个IObservable<IObservable<T>> observable，然后调用Merge如果你明白我的意思的话，就包括新父母的孩子。诀窍是转换ObservableCollection<IObservable<Thing>> to an IObservable<IObservable<Thing>>.

如果您有 ReactiveUI 正在运行，并且可以使用它，那么您可以将ObservableCollection<IObservable<Thing>> to a ReactiveCollection<IObservable<Thing>>. ReactiveCollection继承自ObservableCollection，并且还实现了IObservable.

如果 ReactiveUI 是不可能的（我猜这是因为你已经在使用 Caliburn Micro 集合），那么你可以使用ObservableCollection的事件：

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