Extending the InspectCollectionWithObservable example to nested observable collections

[simonhaines]

I am trying to extend the InspectCollectionWithObservable example to nested collections and have encountered a couple of issues that are likely not bugs, just my misunderstanding of how to implement the pattern.

My goal is to have a root object that has a SourceList of parent objects that each have a SourceList of child objects. Each child object has an observable. When a new value is observed in a child's observable (or children are added and removed to the parent collection), I'd like to produce an enumerable of the latest observed values in each child. Similarly, as parents are added and removed from the root SourceList, I'd like to perform some action with an enumerable of the parent objects and their enumerable of their children's values.

An illustration of my goal:

using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using DynamicData;

var parents = new SourceList<Parent>();
var observers = parents
    .Connect()
    .Transform(p => new ParentObserver(p))
    .DisposeMany()
    .Publish();

var collectionChanged = observers.ToCollection();
var itemChanged = observers
    .MergeMany(p => p.Observable)
    .Select(_ => Unit.Default)
    .StartWith(Unit.Default);

collectionChanged
    .CombineLatest(itemChanged, (items, _) => items)
    .Subscribe(items =>
    {
        Console.WriteLine($"Items: {string.Join(',', items.Select(i => $"[{string.Join(',', i.Values)}]"))}");
    });
using var disposable = observers.Connect();

var parent1 = new Parent();
parent1.AddChild(new Child(1));
parent1.AddChild(new Child(2));
parent1.AddChild(new Child(3));

var parent2 = new Parent();
parent2.AddChild(new Child(4));
parent2.AddChild(new Child(5));
parent2.AddChild(new Child(6));

parents.Add(parent1);
parents.Add(parent2);

var child7 = new Child(7);
parent2.AddChild(child7);  // Throws ArgumentOutOfRangeException
child7.ChangeValue(8);

internal class Parent
{
    public readonly SourceList<Child> Children = new();
    public void AddChild(Child child) => Children.Add(child);
}

internal class Child(int value)
{
    public BehaviorSubject<int> Value { get; } = new(value);
    public void ChangeValue(int value) => Value.OnNext(value);
}

internal class ChildObserver
{
    public ChildObserver(Child child)
        => Observable = child.Value.Do(v => Value = v);

    public readonly IObservable<int> Observable;
    public int Value;
}

internal class ParentObserver : IDisposable
{
    public ParentObserver(Parent parent)
    {
        var observers = parent.Children
            .Connect()
            .Transform(c => new ChildObserver(c))
            .Publish();
            
        var childChanged = observers
            .MergeMany(c => c.Observable)
            .Select(_ => Unit.Default)
            .StartWith(Unit.Default);
        var collectionChanged = observers.ToCollection();
            
        Observable = collectionChanged
            .CombineLatest(childChanged, (items, _) => Values = items.Select(i => i.Value));

        disposable = observers.Connect();
    }

    private readonly IDisposable disposable;
    public readonly IObservable<IEnumerable<int>> Observable;
    public IEnumerable<int> Values = [];
    public void Dispose() => disposable.Dispose();
}

The expected output is:

Items: [1,2,3]
Items: [1,2,3],[4,5,6]
Items: [1,2,3],[4,5,6,7]
Items: [1,2,3],[4,5,6,8]

This code has a couple of problems:

1. As noted in the comment, adding a new child to a parent after it has been added to the root collection throws an ArgumentOutOfRangeException.
2. Even if I omit the three lines where child7 is added to a parent collection and then its value is changed, the cached values in each ChildObserver are never populated (i.e. the child.Value.Do(v => Value = v) code never runs).

What insight have I missed here? Or is there a better way to achieve my goal?
Many thanks.

[simonhaines]

After sleeping on this problem, I've refactored this example to maintain a single cache (instead of distributing it among many ParentObserver and ChildObserver objects) and use SubscribeMany to maintain it.

For others looking to achieve a similar goal, the refactored example is:

using System.Reactive.Disposables;
using System.Reactive.Subjects;
using DynamicData;

var rootObservable =
    new BehaviorSubject<Dictionary<Parent, Dictionary<Child, int>>>(new Dictionary<Parent, Dictionary<Child, int>>());

var parents = new SourceList<Parent>();
parents.Connect().SubscribeMany(p =>
{
    rootObservable.Value[p] = new Dictionary<Child, int>();
    return new CompositeDisposable(
        p.Children.Connect().SubscribeMany(c => new CompositeDisposable(
            c.Value.Subscribe(v =>
            {
                rootObservable.Value[p][c] = v;
                rootObservable.OnNext(rootObservable.Value);
            }),
            Disposable.Create(() =>
            {
                rootObservable.Value[p].Remove(c);
                rootObservable.OnNext(rootObservable.Value);
            }))
        ).Subscribe(),
        Disposable.Create(() =>
        {
            rootObservable.Value.Remove(p);
            rootObservable.OnNext(rootObservable.Value);
        })
    );
}).Subscribe();

rootObservable.Subscribe(cache =>
    Console.WriteLine(string.Concat("Cache: ",
        string.Join(',', cache.Values.Select(v => $"[{string.Join(',', v.Values)}]"))))
);

var parent1 = new Parent();
parent1.AddChild(new Child(1));
parent1.AddChild(new Child(2));
parent1.AddChild(new Child(3));

var parent2 = new Parent();
parent2.AddChild(new Child(4));
parent2.AddChild(new Child(5));
parent2.AddChild(new Child(6));

parents.Add(parent1);
parents.Add(parent2);

var child7 = new Child(7);
parent2.AddChild(child7);
child7.ChangeValue(8);
parents.Remove(parent1);

internal record Parent
{
    public readonly SourceList<Child> Children = new();
    public void AddChild(Child child) => Children.Add(child);
}

internal class Child(int value)
{
    public BehaviorSubject<int> Value { get; } = new(value);
    public void ChangeValue(int value) => Value.OnNext(value);
}

The output is:

Cache: 
Cache: [1]
Cache: [1,2]
Cache: [1,2,3]
Cache: [1,2,3],[4]
Cache: [1,2,3],[4,5]
Cache: [1,2,3],[4,5,6]
Cache: [1,2,3],[4,5,6,7]
Cache: [1,2,3],[4,5,6,8]
Cache: [2,3],[4,5,6,8]
Cache: [3],[4,5,6,8]
Cache: [],[4,5,6,8]
Cache: [4,5,6,8]

I suspect there might be further optimisations available by using a SourceCache as the backing store, but I haven't explored this possibility.

[JakenVeina]

I suspect there might be further optimisations available by using a SourceCache as the backing store, but I haven't explored this possibility.

The thing you'll get with SourceCache<> is the .TransformToTree() operator, which builds the entire "collections within collections" structure for you.

public class ItemViewModel
{
    public long Id { get; init; }
    public long? ParentId { get; init; }
    public string Name { get; init; }
    public IObservable<int> Data { get; }
}

using var items = new SourceCache<ItemViewModel, long>();

var parentItems = items
    .Connect()
    .TransformToTree(static item => item.ParentId)
    .AsObservableCache();
    
using var subscription = Observable.Merge(
        parentItems
            .Connect()
            .Transform(...)
            .ToCollection(),
        parentItems
            .Connect()
            .Transform(static parent => parent.Item.Children)
            .Transform(...)
            .ToCollection())
    .Subscribe(...);

Something to this effect, I think. It's probably not directly applicable to your scenario, but maybe it gives you some new options.