Rewrote the list variant of the OnItemRemoved() operator,...

src/DynamicData.Tests/List/OnItemRemovedFixture.cs

@@ -1,7 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
 using FluentAssertions;
 using Xunit;
 
@@ -11,6 +12,29 @@ namespace DynamicData.Tests.List;
 
 public class OnItemRemovedFixture
 {
+    // https://github.com/reactivemarbles/DynamicData/issues/1061
+    [Theory]
+    [InlineData(true)]
+    [InlineData(false)]
+    public void SubscriberDoesNotHandleErrors_ErrorBubblesUpstream(bool invokeOnUnsubscribe)
+    {
+        using var source = new TestSourceList<int>();
+
+        using var subscription = source.Connect()
+            .OnItemRemoved(
+                removeAction:           static _ => { },
+                invokeOnUnsubscribe:    invokeOnUnsubscribe)
+            .Subscribe();
+
+        var error = new Exception("Test");
+
+        FluentActions.Invoking(() => source.SetError(error))
+            .Should()
+            .Throw<Exception>("errors not handled by the subscriber should propagate upstream to the caller")
+            .Which
+            .Should().BeSameAs(error);
+    }
+
     [Theory]
     [InlineData(0,  0,  0)]
     [InlineData(1,  0,  0)]

src/DynamicData/List/Internal/OnItemRemoved.cs

@@ -0,0 +1,59 @@
+// Copyright (c) 2011-2025 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;
+
+namespace DynamicData.List.Internal;
+
+internal static class OnItemRemoved<T>
+    where T : notnull
+{
+    public static IObservable<IChangeSet<T>> Create(
+        IObservable<IChangeSet<T>> source,
+        Action<T> removeAction,
+        bool invokeOnUnsubscribe)
+    {
+        source.ThrowArgumentNullExceptionIfNull(nameof(source));
+        removeAction.ThrowArgumentNullExceptionIfNull(nameof(removeAction));
+
+        var removalProcessor = source.Do(changeSet =>
+        {
+            foreach (var change in changeSet)
+            {
+                switch (change.Reason)
+                {
+                    case ListChangeReason.Clear:
+                    case ListChangeReason.RemoveRange:
+                        foreach (var item in change.Range)
+                            removeAction.Invoke(item);
+                        break;
+
+                    case ListChangeReason.Remove:
+                        removeAction.Invoke(change.Item.Current);
+                        break;
+
+                    case ListChangeReason.Replace:
+                        removeAction.Invoke(change.Item.Previous.Value);
+                        break;
+                }
+            }
+        });
+
+        return invokeOnUnsubscribe
+            ? Observable.Create<IChangeSet<T>>(observer =>
+            {
+                var items = new List<T>();
+
+                return removalProcessor
+                    .Do(changeSet => items.Clone(changeSet))
+                    .Finally(() =>
+                    {
+                        foreach (var item in items)
+                            removeAction.Invoke(item);
+                    })
+                    .SubscribeSafe(observer);
+            })
+            : removalProcessor;
+    }
+}

src/DynamicData/List/ObservableListEx.cs

@@ -1276,26 +1276,24 @@ public static IObservable<IChangeSet<T>> OnItemRefreshed<T>(
             refreshAction: refreshAction);
 
     /// <summary>
-    /// Callback for each item as and when it is being removed from the stream.
+    /// Invokes a given action for every item removed from the source list stream.
     /// </summary>
-    /// <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="ArgumentNullException">
-    /// source
-    /// or
-    /// removeAction.
-    /// </exception>
-    public static IObservable<IChangeSet<T>> OnItemRemoved<T>(this IObservable<IChangeSet<T>> source, Action<T> removeAction, bool invokeOnUnsubscribe = true)
-        where T : notnull
-    {
-        source.ThrowArgumentNullExceptionIfNull(nameof(source));
-        removeAction.ThrowArgumentNullExceptionIfNull(nameof(removeAction));
-
-        return new OnBeingRemoved<T>(source, removeAction, invokeOnUnsubscribe).Run();
-    }
+    /// <typeparam name="T">The type of items in the list.</typeparam>
+    /// <param name="source">The list stream whose items are to be passed to <paramref name="removeAction"/>.</param>
+    /// <param name="removeAction">The action to invoke upon each removed item.</param>
+    /// <param name="invokeOnUnsubscribe">A flag indicating whether <paramref name="removeAction"/> should be invoked upon teardown of the stream. This includes disposal of subscriptions, completion notifications, and error notifications.</param>
+    /// <returns>A list stream, containing all items in <paramref name="source"/>, with changes published after <paramref name="removeAction"/> has been invoked.</returns>
+    /// <exception cref="ArgumentNullException">Throws for <paramref name="source"/> and <paramref name="removeAction"/>.</exception>
+    /// <remarks>Note that "removed" items includes items from <see cref="ListChangeReason.Remove"/>, <see cref="ListChangeReason.RemoveRange"/>, <see cref="ListChangeReason.Replace"/>, and <see cref="ListChangeReason.Clear"/> changes.</remarks>
+    public static IObservable<IChangeSet<T>> OnItemRemoved<T>(
+                this IObservable<IChangeSet<T>> source,
+                Action<T> removeAction,
+                bool invokeOnUnsubscribe = true)
+            where T : notnull
+        => List.Internal.OnItemRemoved<T>.Create(
+            source: source,
+            removeAction: removeAction,
+            invokeOnUnsubscribe: invokeOnUnsubscribe);
 
     /// <summary>
     /// Apply a logical Or operator between the collections.