Fix OnBeingRemoved and option as to whether OnItemRemoved should be i…

…nvoked upon subscription. Fixes #613 (#616)

src/DynamicData.Tests/Cache/OnItemFixture.cs

@@ -1,7 +1,10 @@
 using System;
-
+using System.Collections.Generic;
+using System.Collections.ObjectModel;
+using System.ComponentModel;
+using DynamicData.Binding;
 using DynamicData.Tests.Domain;
-
+using FluentAssertions;
 using Xunit;
 
 namespace DynamicData.Tests.Cache;
@@ -52,6 +55,21 @@ public void OnItemRemovedCalled()
         Assert.True(called);
     }
 
+    [Fact]
+    [Description("Test for https://github.com/reactivemarbles/DynamicData/issues/613")]
+    public void OnItemRemovedNotCalledForUpdate()
+    {
+        var called = false;
+        var source = new SourceCache<Person, int>(x => x.Age);
+
+        source.Connect().OnItemRemoved(_ => called = true).Subscribe();
+
+        source.AddOrUpdate(new Person("A", 1));
+        source.AddOrUpdate(new Person("A", 2));
+
+        called.Should().Be(false);
+    }
+
     [Fact]
     public void OnItemUpdatedCalled()
     {
@@ -66,4 +84,57 @@ public void OnItemUpdatedCalled()
         source.AddOrUpdate(update);
         Assert.True(called);
     }
+
+    [Fact]
+    [Description("Test for https://github.com/reactivemarbles/DynamicData/issues/268")]
+    public void ListAndCacheShouldHaveEquivalentBehaviour()
+    {
+        var source = new ObservableCollection<Item>
+        {
+            new Item { Id = 1 },
+            new Item { Id = 2 }
+        };
+
+        var list = source.ToObservableChangeSet()
+            .Transform(item => new Proxy { Item = item })
+            .OnItemAdded(proxy => proxy.Active = true)
+            .OnItemRemoved(proxy => proxy.Active = false)
+            .Bind(out var listOutput)
+            .Subscribe();
+
+        var cache = source.ToObservableChangeSet(item => item.Id)
+            .Transform(item => new Proxy { Item = item })
+            .OnItemAdded(proxy => proxy.Active = true)
+            .OnItemRemoved(proxy => proxy.Active = false)
+            .Bind(out var cacheOutput)
+            .Subscribe();
+
+        Assert.Equal(listOutput, cacheOutput, new ProxyEqualityComparer());
+
+        list.Dispose();
+        cache.Dispose();
+
+        Assert.Equal(listOutput, cacheOutput, new ProxyEqualityComparer());
+    }
+
+    public class Item
+    {
+        public int Id { get; set; }
+    }
+
+    public class Proxy
+    {
+        public Item Item { get; set; }
+
+        public bool? Active { get; set; }
+
+
+    }
+
+    public class ProxyEqualityComparer : IEqualityComparer<Proxy>
+    {
+        public bool Equals(Proxy x, Proxy y) => x.Item.Id == y.Item.Id && x.Active == y.Active;
+
+        public int GetHashCode(Proxy obj) => HashCode.Combine(obj.Active, obj.Item);
+    }
 }

src/DynamicData/Cache/Internal/OnBeingRemoved.cs

@@ -2,8 +2,6 @@
 // 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;
-using System.Collections.Generic;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
 
@@ -15,13 +13,15 @@ internal sealed class OnBeingRemoved<TObject, TKey>
     where TKey : notnull
 {
     private readonly Action<TObject> _removeAction;
+    private readonly bool _invokeOnUnsubscribe;
 
     private readonly IObservable<IChangeSet<TObject, TKey>> _source;
 
-    public OnBeingRemoved(IObservable<IChangeSet<TObject, TKey>> source, Action<TObject> removeAction)
+    public OnBeingRemoved(IObservable<IChangeSet<TObject, TKey>> source, Action<TObject> removeAction, bool invokeOnUnsubscribe)
     {
         _source = source ?? throw new ArgumentNullException(nameof(source));
         _removeAction = removeAction ?? throw new ArgumentNullException(nameof(removeAction));
+        _invokeOnUnsubscribe = invokeOnUnsubscribe;
     }
 
     public IObservable<IChangeSet<TObject, TKey>> Run()
@@ -40,7 +40,11 @@ public IObservable<IChangeSet<TObject, TKey>> Run()
 
                         lock (locker)
                         {
-                            cache.Items.ForEach(t => _removeAction(t));
+                            if (_invokeOnUnsubscribe)
+                            {
+                                cache.Items.ForEach(t => _removeAction(t));
+                            }
+
                             cache.Clear();
                         }
                     });
@@ -54,11 +58,6 @@ private void RegisterForRemoval(IChangeSet<TObject, TKey> changes, Cache<TObject
             {
                 switch (change.Reason)
                 {
-                    case ChangeReason.Update:
-                        // ReSharper disable once InconsistentlySynchronizedField
-                        change.Previous.IfHasValue(t => _removeAction(t));
-                        break;
-
                     case ChangeReason.Remove:
                         // ReSharper disable once InconsistentlySynchronizedField
                         _removeAction(change.Current);

src/DynamicData/Cache/ObservableCacheEx.cs

@@ -1129,7 +1129,7 @@ public static IObservable<IChangeSet<TObject, TKey>> DisposeMany<TObject, TKey>(
             throw new ArgumentNullException(nameof(source));
         }
 
-        return new OnBeingRemoved<TObject, TKey>(
+        return new DisposeMany<TObject, TKey>(
             source,
             t =>
             {
@@ -2874,19 +2874,20 @@ public static IObservable<IChangeSet<TObject, TKey>> OnItemRefreshed<TObject, TK
     /// <typeparam name="TKey">The type of the key.</typeparam>
     /// <param name="source">The source.</param>
     /// <param name="removeAction">The remove action.</param>
+    /// <param name="invokeOnUnsubscribe"> Should the remove action be invoked when the subscription is disposed.</param>
     /// <returns>An observable which emits a change set with items being removed.</returns>
     /// <exception cref="System.ArgumentNullException">
     /// source
     /// or
     /// removeAction.
     /// </exception>
-    public static IObservable<IChangeSet<TObject, TKey>> OnItemRemoved<TObject, TKey>(this IObservable<IChangeSet<TObject, TKey>> source, Action<TObject> removeAction)
+    public static IObservable<IChangeSet<TObject, TKey>> OnItemRemoved<TObject, TKey>(this IObservable<IChangeSet<TObject, TKey>> source, Action<TObject> removeAction, bool invokeOnUnsubscribe = true)
         where TKey : notnull
     {
         if (source is null) throw new ArgumentNullException(nameof(source));
         if (removeAction is null) throw new ArgumentNullException(nameof(removeAction));
 
-        return new OnBeingRemoved<TObject, TKey>(source, removeAction).Run();
+        return new OnBeingRemoved<TObject, TKey>(source, removeAction, invokeOnUnsubscribe).Run();
     }
 
     /// <summary>

src/DynamicData/List/Internal/OnBeingRemoved.cs

@@ -14,13 +14,15 @@ namespace DynamicData.List.Internal;
 internal sealed class OnBeingRemoved<T>
 {
     private readonly Action<T> _callback;
+    private readonly bool _invokeOnUnsubscribe;
 
     private readonly IObservable<IChangeSet<T>> _source;
 
-    public OnBeingRemoved(IObservable<IChangeSet<T>> source, Action<T> callback)
+    public OnBeingRemoved(IObservable<IChangeSet<T>> source, Action<T> callback, bool invokeOnUnsubscribe)
     {
         _source = source ?? throw new ArgumentNullException(nameof(source));
         _callback = callback ?? throw new ArgumentNullException(nameof(callback));
+        _invokeOnUnsubscribe = invokeOnUnsubscribe;
     }
 
     public IObservable<IChangeSet<T>> Run()
@@ -36,7 +38,11 @@ public IObservable<IChangeSet<T>> Run()
                     () =>
                     {
                         subscriber.Dispose();
-                        items.ForEach(t => _callback(t));
+
+                        if (_invokeOnUnsubscribe)
+                        {
+                            items.ForEach(t => _callback(t));
+                        }
                     });
             });
     }

src/DynamicData/List/ObservableListEx.cs

@@ -1150,13 +1150,14 @@ public static IObservable<IChangeSet<TObject>> OnItemRefreshed<TObject>(this IOb
     /// <typeparam name="T">The type of the object.</typeparam>
     /// <param name="source">The source.</param>
     /// <param name="removeAction">The remove action.</param>
+    /// <param name="invokeOnUnsubscribe"> Should the remove action be invoked when the subscription is disposed.</param>
     /// <returns>An observable which emits the change set.</returns>
     /// <exception cref="System.ArgumentNullException">
     /// source
     /// or
     /// removeAction.
     /// </exception>
-    public static IObservable<IChangeSet<T>> OnItemRemoved<T>(this IObservable<IChangeSet<T>> source, Action<T> removeAction)
+    public static IObservable<IChangeSet<T>> OnItemRemoved<T>(this IObservable<IChangeSet<T>> source, Action<T> removeAction, bool invokeOnUnsubscribe = true)
     {
         if (source is null)
         {
@@ -1168,7 +1169,7 @@ public static IObservable<IChangeSet<T>> OnItemRemoved<T>(this IObservable<IChan
             throw new ArgumentNullException(nameof(removeAction));
         }
 
-        return new OnBeingRemoved<T>(source, removeAction).Run();
+        return new OnBeingRemoved<T>(source, removeAction, invokeOnUnsubscribe).Run();
     }
 
     /// <summary>