Producing single change set when using composite collection

[Bulgrak]

I have the following interface

public interface ITopic
{
    Guid Id { get; set; }
    string Title { get; set; }
}

And the following classes:

public class TopicA : ITopic
{
    public Guid Id { get; set; }
    public string Title { get; set; }

    public TopicA()
    {
        Id = Guid.NewGuid();
    }
}

public class TopicB : ITopic
{
    public Guid Id { get; set; }
    public string Title { get; set; }

    public TopicB()
    {
        Id = Guid.NewGuid();
    }
}

What I want to achieve is an observable composite collection where collections are added or removed dynamically.
I have tried this with the following code:

var topicAs = new List<ITopic>
{
    new TopicA { Title = "TestA1" },
    new TopicA { Title = "TestA2" }
};

var topicBs = new List<ITopic>
{
    new TopicB { Title = "TestB1" },
    new TopicB { Title = "TestB2" }
};

var topicASourceCache = new SourceCache<ITopic, Guid>(x => x.Id);
topicASourceCache.AddOrUpdate(topicAs);

var topicBSourceCache = new SourceCache<ITopic, Guid>(x => x.Id);
topicBSourceCache.AddOrUpdate(topicBs);

var compositeTopicSourceList = new SourceList<ISourceCache<ITopic, Guid>>();

compositeTopicSourceList.Or().Transform(x => x).Subscribe(x =>
{
    Console.WriteLine("Test");
});

compositeTopicSourceList.Edit(x =>
{
    x.AddRange(new[] { topicASourceCache, topicBSourceCache });
});

topicASourceCache.AddOrUpdate(new TopicA { Title = "TestA3" });

This works, but it produces two change sets when editing compositeTopicSourceList. I want to flatten the list and only have a single change set (to reduce unnecessary overhead).
It almost works if I change compositeTopicSourceList observable to the following:

compositeTopicSourceList.Connect().Transform(x => x).Subscribe(x =>
{
    Console.WriteLine("Test");
});

However, no change set is produced when adding a new topic to the collection topicASourceCache.
Actually, If I use SourceList instead of SourceCache for my composite collection, everything works as I want it to (single change set when editing and change set produced on new topic added):

var topicAs = new List<ITopic>
{
    new TopicA { Title = "TestA1" },
    new TopicA { Title = "TestA2" }
};

var topicBs = new List<ITopic>
{
    new TopicB { Title = "TestB1" },
    new TopicB { Title = "TestB2" }
};

var topicASourceCache = new SourceList<ITopic>();
topicASourceCache.AddRange(topicAs);

var topicBSourceCache = new SourceList<ITopic>();
topicBSourceCache.AddRange(topicBs);

var compositeTopicSourceList = new SourceList<ISourceList<ITopic>>();

compositeTopicSourceList.Connect().TransformMany(x => x).Subscribe(x =>
{
    Console.WriteLine("Test");
});

compositeTopicSourceList.Edit(x =>
{
    x.AddRange(new[] { topicASourceCache, topicBSourceCache });
});

topicASourceCache.Add(new TopicA { Title = "TestA3" });

Any advice on how to achieve a single change set when using composite collection with SourceCache?

[RolandPheasant]

There is no deterministically way to batch the results into a single change set as the code cannot know when

1. each individual source is loaded
2. when the combined list is loaded, or when new lists are added and old lists removed from the composite collection.

This is true for Or and for TransformMany.

However there is a slightly less scientific way which is to use a time based buffer. Something like this:

compositeTopicSourceList.Connect()
      .Or()
     .TransformMany(x => x)
     .Buffer(TimeSpan.FromMilliseconds(150))  // Buffer by time, and experiment with period
     .FlattenBufferResult()  // Flatten the nested result

Sorry for the delay in answering but I do not receive notifications for discussion threads -> there must be a setting some where. I'll take a look,

[Bulgrak]

No worries about the delay. I figured you were just busy with real life. :)
What you say makes sense. However, I think I have achieved what I want more or less with a custom operator using heavy inspiration from the following constructor in DynamicData.List.Internal.TransformMany:

public TransformMany(IObservable<IChangeSet<TSource>> source, Func<TSource, IObservableList<TDestination>> manySelector, IEqualityComparer<TDestination>? equalityComparer = null)
        : this(
            source,
            s => new ManySelectorFunc(s, x => manySelector(x).Items),
            equalityComparer,
            t => Observable.Defer(
                () =>
                {
                    var subsequentChanges = manySelector(t).Connect();

                    if (manySelector(t).Count > 0)
                    {
                        return subsequentChanges;
                    }

                    return Observable.Return(ChangeSet<TDestination>.Empty).Concat(subsequentChanges);
                }))
    {
    }

The custom operator looks like this (basically a mix of DynamicData.Cache.Internal.TransformMany and DynamicData.List.Internal.TransformMany):

using System.Collections;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using DynamicData;
using DynamicData.Kernel;

namespace ConsoleAppDynamicDataCustomOperator;

internal class CacheTransformMany<TDestination, TDestinationKey, TSource, TSourceKey>
    where TSourceKey : notnull
    where TDestinationKey : notnull
{
    private readonly Func<TSource, IObservable<IChangeSet<TDestination, TDestinationKey>>>? _childChanges;

    private readonly Func<TDestination, TDestinationKey> _keySelector;

    private readonly Func<TSource, IEnumerable<TDestination>> _manySelector;

    private readonly IObservable<IChangeSet<TSource, TSourceKey>> _source;

    public CacheTransformMany(IObservable<IChangeSet<TSource, TSourceKey>> source, Func<TSource, IObservableCache<TDestination, TDestinationKey>> manySelector, Func<TDestination, TDestinationKey> keySelector)
        : this(
            source,
            s => new ManySelectorFunc(s, x => manySelector(x).Items),
            keySelector,
            t => Observable.Defer(
                () =>
                {
                    var subsequentChanges = manySelector(t).Connect();
                    if (manySelector(t).Count > 0)
                    {
                        return subsequentChanges;
                    }

                    return Observable.Return(ChangeSet<TDestination, TDestinationKey>.Empty).Concat(subsequentChanges);
                }))
    {
    }

    public CacheTransformMany(IObservable<IChangeSet<TSource, TSourceKey>> source, Func<TSource, IEnumerable<TDestination>> manySelector, Func<TDestination, TDestinationKey> keySelector, Func<TSource, IObservable<IChangeSet<TDestination, TDestinationKey>>>? childChanges = null)
    {
        _source = source;
        _manySelector = manySelector;
        _keySelector = keySelector;
        _childChanges = childChanges;
    }

    public IObservable<IChangeSet<TDestination, TDestinationKey>> Run()
    {
        return _childChanges is null ? Create() : CreateWithChangeSet();
    }

    private IObservable<IChangeSet<TDestination, TDestinationKey>> Create()
    {
        return _source.Transform(
            (t, _) =>
            {
                var destination = _manySelector(t).Select(m => new DestinationContainer(m, _keySelector(m))).ToArray();
                return new ManyContainer(() => destination);
            },
            true).Select(changes => new ChangeSet<TDestination, TDestinationKey>(new DestinationEnumerator(changes)));
    }

    private IObservable<IChangeSet<TDestination, TDestinationKey>> CreateWithChangeSet()
    {
        if (_childChanges is null)
        {
            throw new InvalidOperationException("The childChanges is null and should not be.");
        }

        return Observable.Create<IChangeSet<TDestination, TDestinationKey>>(
            observer =>
            {
                var result = new ChangeAwareCache<TDestination, TDestinationKey>();

                var transformed = _source.Transform(
                    (t, _) =>
                    {
                        // Only skip initial for first time Adds where there is initial data records
                        var locker = new object();
                        var collection = _manySelector(t);
                        var changes = _childChanges(t).Synchronize(locker).Skip(1);
                        return new ManyContainer(
                            () =>
                            {
                                lock (locker)
                                {
                                    return collection.Select(m => new DestinationContainer(m, _keySelector(m))).ToArray();
                                }
                            },
                            changes);
                    }).Publish();

                var outerLock = new object();
                var initial = transformed.Synchronize(outerLock).Select(changes => {
                    return new ChangeSet<TDestination, TDestinationKey>(new DestinationEnumerator(changes));
                });

                transformed.Subscribe(x =>
                {

                });

                var subsequent = transformed.MergeMany(x => {
                    return x.Changes;
                }).Synchronize(outerLock);

                var allChanges = initial.Merge(subsequent).Select(
                    changes =>
                    {
                        result.Clone(changes);
                        return result.CaptureChanges();
                    });

                return new CompositeDisposable(allChanges.SubscribeSafe(observer), transformed.Connect());
            });
    }

    private sealed class DestinationContainer
    {
        public DestinationContainer(TDestination item, TDestinationKey key)
        {
            Item = item;
            Key = key;
        }

        public static IEqualityComparer<DestinationContainer> KeyComparer { get; } = new KeyEqualityComparer();

        public TDestination Item { get; }

        public TDestinationKey Key { get; }

        private sealed class KeyEqualityComparer : IEqualityComparer<DestinationContainer>
        {
            public bool Equals(DestinationContainer? x, DestinationContainer? y)
            {
                if (x is null && y is null)
                {
                    return true;
                }

                if (x is null || y is null)
                {
                    return false;
                }

                return EqualityComparer<TDestinationKey?>.Default.Equals(x.Key, y.Key);
            }

            public int GetHashCode(DestinationContainer obj)
            {
                return EqualityComparer<TDestinationKey?>.Default.GetHashCode(obj.Key);
            }
        }
    }

    private sealed class DestinationEnumerator : IEnumerable<Change<TDestination, TDestinationKey>>
    {
        private readonly IChangeSet<ManyContainer, TSourceKey> _changes;

        public DestinationEnumerator(IChangeSet<ManyContainer, TSourceKey> changes)
        {
            _changes = changes;
        }

        public IEnumerator<Change<TDestination, TDestinationKey>> GetEnumerator()
        {
            foreach (var change in _changes)
            {
                switch (change.Reason)
                {
                    case ChangeReason.Add:
                    case ChangeReason.Remove:
                    case ChangeReason.Refresh:
                    {
                        foreach (var destination in change.Current.Destination)
                        {
                            yield return new Change<TDestination, TDestinationKey>(change.Reason, destination.Key, destination.Item);
                        }
                    }

                        break;
                    case ChangeReason.Update:
                    {
                        var previousItems = change.Previous.Value.Destination.AsArray();
                        var currentItems = change.Current.Destination.AsArray();

                        var removes = previousItems.Except(currentItems, DestinationContainer.KeyComparer);
                        var adds = currentItems.Except(previousItems, DestinationContainer.KeyComparer);
                        var updates = currentItems.Intersect(previousItems, DestinationContainer.KeyComparer);

                        foreach (var destination in removes)
                        {
                            yield return new Change<TDestination, TDestinationKey>(ChangeReason.Remove, destination.Key, destination.Item);
                        }

                        foreach (var destination in adds)
                        {
                            yield return new Change<TDestination, TDestinationKey>(ChangeReason.Add, destination.Key, destination.Item);
                        }

                        foreach (var destination in updates)
                        {
                            var current = currentItems.First(d => d.Key.Equals(destination.Key));
                            var previous = previousItems.First(d => d.Key.Equals(destination.Key));

                            // Do not update is items are the same reference
                            if (!ReferenceEquals(current.Item, previous.Item))
                            {
                                yield return new Change<TDestination, TDestinationKey>(ChangeReason.Update, destination.Key, current.Item, previous.Item);
                            }
                        }
                    }

                        break;
                }
            }
        }

        IEnumerator IEnumerable.GetEnumerator()
        {
            return GetEnumerator();
        }
    }

    private sealed class ManyContainer
    {
        private readonly Func<IEnumerable<DestinationContainer>> _initial;

        public ManyContainer(Func<IEnumerable<DestinationContainer>> initial, IObservable<IChangeSet<TDestination, TDestinationKey>>? changes = null)
        {
            _initial = initial;
            Changes = changes ?? Observable.Empty<IChangeSet<TDestination, TDestinationKey>>();
        }
            
        public IObservable<IChangeSet<TDestination, TDestinationKey>> Changes { get; }

        public IEnumerable<DestinationContainer> Destination => _initial();
    }
        
    private class ManySelectorFunc : IEnumerable<TDestination>
    {
        private readonly TSource _source;

        private readonly Func<TSource, IEnumerable<TDestination>> _selector;

        public ManySelectorFunc(TSource source, Func<TSource, IEnumerable<TDestination>> selector)
        {
            _source = source;
            _selector = selector ?? throw new ArgumentNullException(nameof(selector));
        }

        public IEnumerator<TDestination> GetEnumerator() => _selector(_source).GetEnumerator();

        IEnumerator IEnumerable.GetEnumerator() => _selector(_source).GetEnumerator();
    }
}

Here is the extension method for the custom operator:

using DynamicData;

namespace ConsoleAppDynamicDataCustomOperator;

public static class DynamicDataExtensions
{
    public static IObservable<IChangeSet<TDestination, TDestinationKey>> CacheTransformMany<TDestination, TDestinationKey, TSource, TSourceKey>(this IObservable<IChangeSet<TSource, TSourceKey>> source, Func<TSource, IObservableCache<TDestination, TDestinationKey>> manySelector, Func<TDestination, TDestinationKey> keySelector)
    {
        return new CacheTransformMany<TDestination, TDestinationKey, TSource, TSourceKey>(source, manySelector, keySelector).Run();
    }
}

And the demo class:

using DynamicData;

namespace ConsoleAppDynamicDataCustomOperator;

public static class SourceCacheDemo
{
    public static void Run()
    {
        var foobars1SourceCache = new SourceCache<IFoobar, Guid>(x => x.Id);
        var foobar1 = new Foobar(1);
        foobars1SourceCache.AddOrUpdate(foobar1);

        var foobars2SourceCache = new SourceCache<IFoobar, Guid>(x => x.Id);
        var foobar2 = new Foobar(2);
        foobars2SourceCache.AddOrUpdate(foobar2);

        var compositeFoobarSourceList = new SourceCache<ISourceCache<IFoobar, Guid>, object>(o => o);
        compositeFoobarSourceList.Connect().CacheTransformMany(x => x, x => x.Id).Subscribe(changes =>
        {
            Console.WriteLine("Start outputting emitted changes");
            foreach (var change in changes)
            {
                Console.WriteLine($"Change reason: {change.Reason}. Foobar number: {change.Current.Number}");
            }
            Console.WriteLine("End outputting emitted changes");
        });

        Console.WriteLine("Start with loading foobars1SourceCache containing foobar 1 and foobars2SourceCache containing foobar 2 into compositeFoobarSourceList");

        compositeFoobarSourceList.Edit(x =>
        {
            x.AddOrUpdate(foobars1SourceCache);
            x.AddOrUpdate(foobars2SourceCache);
        });

        Console.WriteLine("\nPress any key to add foobar 3 and then 4");
        Console.ReadKey();
        var foobar3 = new Foobar(3);
        foobars1SourceCache.AddOrUpdate(foobar3);
        var foobar4 = new Foobar(4);
        foobars2SourceCache.AddOrUpdate(foobar4);

        Console.WriteLine("\nPress any key to remove foobar 3");
        Console.ReadKey();
        foobars1SourceCache.Remove(foobar3);
            
        Console.WriteLine("\nPress any key to add foobars3SourceCache containing foobar 5 and 6");
        Console.ReadKey();
        var foobars3SourceCache = new SourceCache<IFoobar, Guid>(x => x.Id);
        var foobar5 = new Foobar(5);
        var foobar6 = new Foobar(6);
        foobars3SourceCache.AddOrUpdate(foobar5);
        foobars3SourceCache.AddOrUpdate(foobar6);
        compositeFoobarSourceList.AddOrUpdate(foobars3SourceCache);
            
        Console.WriteLine("\nPress any key to add foobar 7 and 8 at the same time");
        Console.ReadKey();
        var foobar7 = new Foobar(7);
        var foobar8 = new Foobar(8);
        foobars3SourceCache.AddOrUpdate(new [] { foobar7, foobar8 });
        
        Console.WriteLine("\nPress any key to remove foobar 7 and 8 at the same time");
        Console.ReadKey();
        foobars3SourceCache.Remove(new []{ foobar7, foobar8 });
        
        Console.WriteLine("\nPress any key to clear compositeFoobarSourceList");
        Console.ReadKey();
        compositeFoobarSourceList.Clear();
        Console.ReadLine();
    }
    
    
    public class Foobar : IFoobar
    {
        public Guid Id { get; } = Guid.NewGuid();
        public int Number { get; }

        public Foobar(int number)
        {
            Number = number;
        }
    }
    
    public interface IFoobar
    {
        public Guid Id { get; }
        public int Number { get; }
    }
}

The demo produces the following output:

Start with loading foobars1SourceCache containing foobar 1 and foobars2SourceCache containing foobar 2 into compositeFoobarSourceList
Start outputting emitted changes
Change reason: Add. Foobar number: 1
Change reason: Add. Foobar number: 2
End outputting emitted changes

Press any key to add foobar 3 and then 4
Start outputting emitted changes
Change reason: Add. Foobar number: 3
End outputting emitted changes
Start outputting emitted changes
Change reason: Add. Foobar number: 4
End outputting emitted changes

Press any key to remove foobar 3
Start outputting emitted changes
Change reason: Remove. Foobar number: 3
End outputting emitted changes

Press any key to add foobars3SourceCache containing foobar 5 and 6
Start outputting emitted changes
Change reason: Add. Foobar number: 5
Change reason: Add. Foobar number: 6
End outputting emitted changes

Press any key to add foobar 7 and 8 at the same time
Start outputting emitted changes
Change reason: Add. Foobar number: 7
Change reason: Add. Foobar number: 8
End outputting emitted changes

Press any key to remove foobar 7 and 8 at the same time
Start outputting emitted changes
Change reason: Remove. Foobar number: 7
Change reason: Remove. Foobar number: 8
End outputting emitted changes

Press any key to clear compositeFoobarSourceList
Start outputting emitted changes
Change reason: Remove. Foobar number: 1
Change reason: Remove. Foobar number: 2
Change reason: Remove. Foobar number: 4
Change reason: Remove. Foobar number: 5
Change reason: Remove. Foobar number: 6
End outputting emitted changes

I haven't tested this extensively, but so far it seems to do what I want to achieve. When multiple sources are loaded into the composite collection, a single changeset is produced. Same applies when all sources are cleared from composite collection.
Do you think this is a valid approach?
(I'm pretty new to both DynamicData and Reactive Extensions, so sorry if I'm using the wrong terms. Still learning. :))