合并不断变化的可观察对象集合答案

【问题标题】：Merging a changing collection of observables合并不断变化的可观察对象集合
【发布时间】：2018-03-05 11:04:56
【问题描述】：

我们有一个实现IObservable<Thing> 的类Thing。在另一个类中，有一个 Things 的集合，该类需要以统一的方式对所有这些可观察对象的更新做出反应。显而易见的方法是Observable.Merge()，这通常有效；然而，当集合发生变化时，我们还需要在合并的订阅中订阅任何新的Things（理论上取消订阅所有已删除的，但这似乎不太成问题 - 他们刚刚赢了'不再产生任何更新）。

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

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

【问题讨论】：

持有 Thing Observables 的集合类型是什么？
@Shlomo 哦，我应该提到这一点。这是一个ObservableCollection（更准确地说是一个 Caliburn.Micro BindableCollection）。
我还假设您没有class Thing : IObservable<Thing>，这将是相当迷幻的。错字？也许你的意思是class Thing : IObservable<Thing2>？
不，其实是IObservable<Thing>；每个实例只是通知订阅者 发生了一些事情，这样以后可以在合并的 Observable 中再次识别实例。
我建议不要实施IObservable<T>。我会进一步建议不要做class Thing : IObservable<Thing>。对您的开发团队成员施加的惩罚似乎是残酷和不寻常的惩罚。

标签： system.reactive

【解决方案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 的事件进行转换：

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

【讨论】：