开启 IObservable<IEnumerable<IEnumerable<T>>>

Question

我有以下结构：

// source of data
interface IItem
{
    IObservable<string> Changed { get; }
}

interface IItemCollection
{
    List<IItem> Items { get; }
    IObservable<IItem> ItemAdded { get; }
    IObservable<IItem> ItemRemoved { get; }
}

interface IItemCollectionManager
{
    List<IItemCollection> ItemCollectionCollection { get; }
    IObservable<IItemCollection> ItemCollectionAdded { get; }
    IObservable<IItemCollection> ItemCollectionRemoved { get; }
}

// desired result
interface IAggregation
{
    IObservable<string> Changed { get; }
}

这里的目标是让IAggregation 公开单个可观察对象。但是，IItems 可以随时从每个IItemCollection 添加和删除，事实上，IItemCollection 也可以随时从IItemCollectionManager 添加或删除。当然，当添加了这样的IItemCollection 时，Aggregation 也应该从那个中发出值，如果删除了 ItemCollection，我不再需要该集合中 IItems 中的 strings被发射。此外，当Item 被添加到任何IItemCollection 时，来自其Changed observable 的值也应该产生来自IAggregation 的Changed observable 的值。

现在，当只有一个 IItemCollection 时，解决这个问题相当简单，例如像这样：

class AggregationImpl : IAggregation 
{
    public AggregationImpl(IItemCollection itemCollection)
    {
        var added = itemCollection.ItemAdded
            .Select(_ => itemCollection.Items);
        var removed = itemCollection.ItemRemoved
            .Select(_ => itemCollection.Items);

        Changed = Observable.Merge(added, removed)
            .StartWith(itemCollection.Items)
            .Select(coll => coll.Select(item => item.Changed).Merge())
            .Switch();
    }

    public IObservable<string> Changed { get; }

}

...这里的关键点是我将所有Item 的Changed 可观察对象扁平化为带有Merge() 的单个可观察对象，然后，每次添加或删除项目时，我都会重新创建整个Observable 并使用Switch() 退订旧的并订阅新的`。

我觉得扩展以包含 IItemCollectionManager 应该很简单，但我不太确定如何处理它。

Answer 1

我希望这行得通，或者至少让你走上正确的道路。由于测试似乎相当复杂，因此我将对其进行改进。如果你有一些简单的测试代码，那么我很乐意测试。

首先，我不太喜欢您发布的实现。您正在连接到ItemAdded 和ItemRemoved obeservables，根本不使用数据；您正在从 Items 属性中获取数据。这可能会在糟糕的实现中导致竞争条件，即在属性更新之前发送事件。因此，我创建了自己的实现。我还建议将其放入扩展方法中，因为这样可以让以后的生活更轻松：

public static IObservable<string> ToAggregatedObservable(this IItemCollection itemCollection)
{
    return Observable.Merge(
            itemCollection.ItemAdded.Select(item => (op: "+", item)),
            itemCollection.ItemRemoved.Select(item => (op: "-", item))
        )
        .Scan(ImmutableList<IItem>.Empty.AddRange(itemCollection.Items), (list, t) =>
            t.op == "+"
                ? list.Add(t.item)
                : list.Remove(t.item)
        )
        .Select(l => l.Select(item => item.Changed).Merge())
        .Switch();

}

请暂时原谅魔术字符串，如果您愿意，可以将其变成enum。我们在Scan 内部的ImmutableList 中维护当前项目的状态。当一个项目被添加/删除时，我们更新列表，然后切换 observable。

同样的逻辑可以应用到集合管理器级别：

public static IObservable<string> ToAggregatedObservable(this IItemCollectionManager itemCollectionManager)
{
    return Observable.Merge(
            itemCollectionManager.ItemCollectionAdded.Select(itemColl => (op: "+", itemColl)),
            itemCollectionManager.ItemCollectionRemoved.Select(itemColl => (op: "-", itemColl))
        )
        .Scan(ImmutableList<IItemCollection>.Empty.AddRange(itemCollectionManager.ItemCollectionCollection), (list, t) =>
            t.op == "+"
                ? list.Add(t.itemColl)
                : list.Remove(t.itemColl)
        )
        .Select(l => l.Select(itemColl => itemColl.ToAggregatedObservable()).Merge())
        .Switch();
}

这里我们只是重用了第一种扩展方法，并使用与以前相同的添加/删除然后切换逻辑。