合并不断变化的可观测数据集合 [英] Merging a changing collection of observables

问题描述

我们有一个实现IObservable<Thing>的类Thing。在另一个类中，有一个Thing的集合，该类需要以统一的方式对来自所有这些可观测对象的更新做出反应。最明显的方法是Observable.Merge()，这通常是有效的；然而，当集合发生变化时，我们还需要订阅合并订阅中的任何新的Thing(理论上还需要取消订阅所有已删除的订阅，但这似乎没什么问题--它们只是不再生成任何更新)。

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

合并像这样不断变化的可观测集合的正确方式是什么？

推荐答案

如果您有IObservable<IObservable<T>> observable，则对其调用Merge将包括新父母的孩子，如果您明白我的意思的话。诀窍是将ObservableCollection<IObservable<Thing>>转换为IObservable<IObservable<Thing>>。

如果您到处运行Reactive UI，并且可以使用它，那么您可以将ObservableCollection<IObservable<Thing>>转换为ReactiveCollection<IObservable<Thing>>。ReactiveCollection继承自ObservableCollection，也实现IObservable。

如果Reactive UI是不可能的(我猜这是因为您已经在使用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; }
}

这篇关于合并不断变化的可观测数据集合的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！