Feature: EditDiff extension method for IObservable<Optional<T>>

src/DynamicData.Tests/Cache/EditDiffChangeSetOptionalFixture.cs

@@ -0,0 +1,231 @@
+using System;
+using System.Collections;
+using System.Collections.Generic;
+using System.ComponentModel;
+using System.Diagnostics.CodeAnalysis;
+using System.Linq;
+using System.Reactive.Linq;
+using DynamicData.Kernel;
+using FluentAssertions;
+
+using Xunit;
+
+namespace DynamicData.Tests.Cache;
+
+public class EditDiffChangeSetOptionalFixture
+{
+    private static readonly Optional<Person> s_noPerson = Optional.None<Person>();
+
+    private const int MaxItems = 1097;
+
+    [Fact]
+    [Description("Required to maintain test coverage percentage")]
+    public void NullChecksArePerformed()
+    {
+        Action actionNullKeySelector = () => Observable.Empty<Optional<Person>>().EditDiff<Person, int>(null!);
+        Action actionNullObservable = () => default(IObservable<Optional<Person>>)!.EditDiff<Person, int>(null!);
+
+        actionNullKeySelector.Should().Throw<ArgumentNullException>().WithParameterName("keySelector");
+        actionNullObservable.Should().Throw<ArgumentNullException>().WithParameterName("source");
+    }
+
+    [Fact]
+    public void OptionalSomeCreatesAddChange()
+    {
+        // having
+        var optional = CreatePerson(0, "Name");
+        var optObservable = Observable.Return(optional);
+
+        // when
+        var observableChangeSet = optObservable.EditDiff(p => p.Id);
+        using var results = observableChangeSet.AsAggregator();
+
+        // then
+        results.Data.Count.Should().Be(1);
+        results.Messages.Count.Should().Be(1);
+    }
+
+    [Fact]
+    public void OptionalNoneCreatesRemoveChange()
+    {
+        // having
+        var optional = CreatePerson(0, "Name");
+        var optObservable = new[] {optional, s_noPerson}.ToObservable();
+
+        // when
+        var observableChangeSet = optObservable.EditDiff(p => p.Id);
+        using var results = observableChangeSet.AsAggregator();
+
+        // then
+        results.Data.Count.Should().Be(0);
+        results.Messages.Count.Should().Be(2);
+        results.Messages[0].Adds.Should().Be(1);
+        results.Messages[1].Removes.Should().Be(1);
+        results.Messages[1].Updates.Should().Be(0);
+    }
+
+    [Fact]
+    public void OptionalSomeWithSameKeyCreatesUpdateChange()
+    {
+        // having
+        var optional1 = CreatePerson(0, "Name");
+        var optional2 = CreatePerson(0, "Update");
+        var optObservable = new[] { optional1, optional2 }.ToObservable();
+
+        // when
+        var observableChangeSet = optObservable.EditDiff(p => p.Id);
+        using var results = observableChangeSet.AsAggregator();
+
+        // then
+        results.Data.Count.Should().Be(1);
+        results.Messages.Count.Should().Be(2);
+        results.Messages[0].Adds.Should().Be(1);
+        results.Messages[1].Removes.Should().Be(0);
+        results.Messages[1].Updates.Should().Be(1);
+    }
+
+    [Fact]
+    public void OptionalSomeWithSameReferenceCreatesNoChanges()
+    {
+        // having
+        var optional = CreatePerson(0, "Name");
+        var optObservable = new[] { optional, optional }.ToObservable();
+
+        // when
+        var observableChangeSet = optObservable.EditDiff(p => p.Id);
+        using var results = observableChangeSet.AsAggregator();
+
+        // then
+        results.Data.Count.Should().Be(1);
+        results.Messages.Count.Should().Be(1);
+        results.Summary.Overall.Adds.Should().Be(1);
+        results.Summary.Overall.Removes.Should().Be(0);
+        results.Summary.Overall.Updates.Should().Be(0);
+    }
+
+    [Fact]
+    public void OptionalSomeWithSameCreatesNoChanges()
+    {
+        // having
+        var optional1 = CreatePerson(0, "Name");
+        var optional2 = CreatePerson(0, "Name");
+        var optObservable = new[] { optional1, optional2 }.ToObservable();
+
+        // when
+        var observableChangeSet = optObservable.EditDiff(p => p.Id, new PersonComparer());
+        using var results = observableChangeSet.AsAggregator();
+
+        // then
+        results.Data.Count.Should().Be(1);
+        results.Messages.Count.Should().Be(1);
+        results.Summary.Overall.Adds.Should().Be(1);
+        results.Summary.Overall.Removes.Should().Be(0);
+        results.Summary.Overall.Updates.Should().Be(0);
+    }
+
+    [Fact]
+    public void OptionalSomeWithDifferentKeyCreatesAddRemoveChanges()
+    {
+        // having
+        var optional1 = CreatePerson(0, "Name");
+        var optional2 = CreatePerson(1, "Update");
+        var optObservable = new[] { optional1, optional2 }.ToObservable();
+
+        // when
+        var observableChangeSet = optObservable.EditDiff(p => p.Id);
+        using var results = observableChangeSet.AsAggregator();
+
+        // then
+        results.Data.Count.Should().Be(1);
+        results.Messages.Count.Should().Be(2);
+        results.Messages[0].Adds.Should().Be(1);
+        results.Messages[1].Removes.Should().Be(1);
+        results.Messages[1].Updates.Should().Be(0);
+    }
+    [Theory]
+    [InlineData(true)]
+    [InlineData(false)]
+    public void ResultCompletesIfAndOnlyIfSourceCompletes(bool completeSource)
+    {
+        // having
+        var optional = CreatePerson(0, "Name");
+        var optObservable = Observable.Return(optional);
+        if (!completeSource)
+        {
+            optObservable = optObservable.Concat(Observable.Never<Optional<Person>>());
+        }
+        bool completed = false;
+
+        // when
+        using var results = optObservable.Subscribe(_ => { }, () => completed = true);
+
+        // then
+        completed.Should().Be(completeSource);
+    }
+
+    [Theory]
+    [InlineData(true)]
+    [InlineData(false)]
+    public void ResultFailsIfAndOnlyIfSourceFails (bool failSource)
+    {
+        // having
+        var optional = CreatePerson(0, "Name");
+        var optObservable = Observable.Return(optional);
+        var testException = new Exception("Test");
+        if (failSource)
+        {
+            optObservable = optObservable.Concat(Observable.Throw<Optional<Person>>(testException));
+        }
+        var receivedError = default(Exception);
+
+        // when
+        using var results = optObservable.Subscribe(_ => { }, err => receivedError = err);
+
+        // then
+        receivedError.Should().Be(failSource ? testException : default);
+    }
+
+    [Trait("Performance", "Manual run only")]
+    [Theory]
+    [InlineData(7)]
+    [InlineData(MaxItems)]
+    public void Perf(int maxItems)
+    {
+        // having
+        var optionals = Enumerable.Range(0, maxItems).Select(n => (n % 2) == 0 ? CreatePerson(n, "Name") : s_noPerson);
+        var optObservable = optionals.ToObservable();
+
+        // when
+        var observableChangeSet = optObservable.EditDiff(p => p.Id);
+        using var results = observableChangeSet.AsAggregator();
+
+        // then
+        results.Data.Count.Should().Be(1);
+        results.Messages.Count.Should().Be(maxItems);
+        results.Summary.Overall.Adds.Should().Be((maxItems / 2) + ((maxItems % 2) == 0 ? 0 : 1));
+        results.Summary.Overall.Removes.Should().Be(maxItems / 2);
+        results.Summary.Overall.Updates.Should().Be(0);
+    }
+
+    private static Optional<Person> CreatePerson(int id, string name) => Optional.Some(new Person(id, name));
+
+    private class PersonComparer : IEqualityComparer<Person>
+    {
+        public bool Equals([DisallowNull] Person x, [DisallowNull] Person y) =>
+            EqualityComparer<string>.Default.Equals(x.Name, y.Name) && EqualityComparer<int>.Default.Equals(x.Id, y.Id);
+        public int GetHashCode([DisallowNull] Person obj) => throw new NotImplementedException();
+    }
+
+    private class Person
+    {
+        public Person(int id, string name)
+        {
+            Id = id;
+            Name = name;
+        }
+
+        public int Id { get; }
+
+        public string Name { get; }
+    }
+}

src/DynamicData/Cache/Internal/EditDiffChangeSetOptional.cs

@@ -0,0 +1,92 @@
+// Copyright (c) 2011-2023 Roland Pheasant. All rights reserved.
+// Roland Pheasant licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for full license information.
+
+using System.Reactive.Linq;
+using DynamicData.Kernel;
+
+namespace DynamicData.Cache.Internal;
+
+internal sealed class EditDiffChangeSetOptional<TObject, TKey>
+    where TObject : notnull
+    where TKey : notnull
+{
+    private readonly IObservable<Optional<TObject>> _source;
+
+    private readonly IEqualityComparer<TObject> _equalityComparer;
+
+    private readonly Func<TObject, TKey> _keySelector;
+
+    public EditDiffChangeSetOptional(IObservable<Optional<TObject>> source, Func<TObject, TKey> keySelector, IEqualityComparer<TObject>? equalityComparer)
+    {
+        _source = source ?? throw new ArgumentNullException(nameof(source));
+        _keySelector = keySelector ?? throw new ArgumentNullException(nameof(keySelector));
+        _equalityComparer = equalityComparer ?? EqualityComparer<TObject>.Default;
+    }
+
+    public IObservable<IChangeSet<TObject, TKey>> Run()
+    {
+        return Observable.Create<IChangeSet<TObject, TKey>>(observer =>
+        {
+            var previous = Optional.None<ValueContainer>();
+
+            return _source.Synchronize().Subscribe(
+                nextValue =>
+                {
+                    var current = nextValue.Convert(val => new ValueContainer(val, _keySelector(val)));
+
+                    // Determine the changes
+                    var changes = (previous.HasValue, current.HasValue) switch
+                    {
+                        (true, true) => CreateUpdateChanges(previous.Value, current.Value),
+                        (false, true) => new[] { new Change<TObject, TKey>(ChangeReason.Add, current.Value.Key, current.Value.Object) },
+                        (true, false) => new[] { new Change<TObject, TKey>(ChangeReason.Remove, previous.Value.Key, previous.Value.Object) },
+                        (false, false) => Array.Empty<Change<TObject, TKey>>(),
+                    };
+
+                    // Save the value for the next round
+                    previous = current;
+
+                    // If there are changes, emit as a ChangeSet
+                    if (changes.Length > 0)
+                    {
+                        observer.OnNext(new ChangeSet<TObject, TKey>(changes));
+                    }
+                }, observer.OnError, observer.OnCompleted);
+        });
+    }
+
+    private Change<TObject, TKey>[] CreateUpdateChanges(in ValueContainer prev, in ValueContainer curr)
+    {
+        if (EqualityComparer<TKey>.Default.Equals(prev.Key, curr.Key))
+        {
+            // Key is the same, so Update (unless values are equal)
+            if (!_equalityComparer.Equals(prev.Object, curr.Object))
+            {
+                return new[] { new Change<TObject, TKey>(ChangeReason.Update, curr.Key, curr.Object, prev.Object) };
+            }
+
+            return Array.Empty<Change<TObject, TKey>>();
+        }
+
+        // Key Change means Remove/Add
+        return new[]
+        {
+            new Change<TObject, TKey>(ChangeReason.Remove, prev.Key, prev.Object),
+            new Change<TObject, TKey>(ChangeReason.Add, curr.Key, curr.Object)
+        };
+    }
+
+    private readonly struct ValueContainer
+    {
+        public ValueContainer(TObject obj, TKey key)
+        {
+            Object = obj;
+            Key = key;
+        }
+
+        public TObject Object { get; }
+
+        public TKey Key { get; }
+    }
+}

src/DynamicData/Cache/ObservableCacheEx.cs

@@ -1308,6 +1308,33 @@
         return new EditDiffChangeSet<TObject, TKey>(source, keySelector, equalityComparer).Run();
     }
 
+    /// <summary>
+    /// Converts an Observable Optional to an Observable ChangeSet that adds/removes/updates as the optional changes.
+    /// </summary>
+    /// <typeparam name="TObject">The type of the object.</typeparam>
+    /// <typeparam name="TKey">The type of the key.</typeparam>
+    /// <param name="source">The source.</param>
+    /// <param name="keySelector">Key Selection Function for the ChangeSet.</param>
+    /// <param name="equalityComparer">Optional <see cref="IEqualityComparer{T}"/> instance to use for comparing values.</param>
+    /// <returns>An observable changeset.</returns>
+    /// <exception cref="System.ArgumentNullException">source.</exception>
+    public static IObservable<IChangeSet<TObject, TKey>> EditDiff<TObject, TKey>(this IObservable<Optional<TObject>> source, Func<TObject, TKey> keySelector, IEqualityComparer<TObject>? equalityComparer = null)
+        where TObject : notnull
+        where TKey : notnull
+    {
+        if (source is null)
+        {
+            throw new ArgumentNullException(nameof(source));
+        }
+
+        if (keySelector is null)
+        {
+            throw new ArgumentNullException(nameof(keySelector));
+        }
+
+        return new EditDiffChangeSetOptional<TObject, TKey>(source, keySelector, equalityComparer).Run();
+    }
+
     /// <summary>
     /// Signal observers to re-evaluate the specified item.
     /// </summary>
@@ -4528,7 +4555,7 @@
             return optional.Merge(missing);
         }
 
         return source.ToObservableOptional(key, equalityComparer);
     }
 
     /// <summary>