diff --git a/DynamicData.Tests/Cache/OnItemFixture.cs b/DynamicData.Tests/Cache/OnItemFixture.cs
new file mode 100644
index 000000000..73a196761
--- /dev/null
+++ b/DynamicData.Tests/Cache/OnItemFixture.cs
@@ -0,0 +1,58 @@
+using System;
+using DynamicData.Tests.Domain;
+using Xunit;
+
+namespace DynamicData.Tests.Cache
+{
+    public class OnItemFixture
+    {
+        [Fact]
+        public void OnItemAddCalled()
+        {
+            var called = false;
+            var source = new SourceCache<Person, int>(x => x.Age);
+
+            source.Connect()
+                .OnItemAdded(_ => called = true)
+                .Subscribe();
+
+            var person = new Person("A", 1);
+            
+            source.AddOrUpdate(person);
+            Assert.True(called);
+        }
+
+        [Fact]
+        public void OnItemRemovedCalled()
+        {
+            var called = false;
+            var source = new SourceCache<Person, int>(x => x.Age);
+
+            source.Connect()
+                .OnItemRemoved(_ => called = true)
+                .Subscribe();
+
+            var person = new Person("A", 1);
+            source.AddOrUpdate(person);
+            source.Remove(person);
+            Assert.True(called);
+        }
+
+        [Fact]
+        public void OnItemUpdatedCalled()
+        {
+            var called = false;
+            var source = new SourceCache<Person, int>(x => x.Age);
+
+            source.Connect()
+                .OnItemUpdated((x,y) => called = true)
+                .Subscribe();
+
+            var person = new Person("A", 1);
+            source.AddOrUpdate(person);
+            var update = new Person("B", 1);
+            source.AddOrUpdate(update);
+            Assert.True(called);
+        }
+    }
+}
\ No newline at end of file
diff --git a/DynamicData/Cache/Internal/OnItemRemoved.cs b/DynamicData/Cache/Internal/DisposeMany.cs
similarity index 91%
rename from DynamicData/Cache/Internal/OnItemRemoved.cs
rename to DynamicData/Cache/Internal/DisposeMany.cs
index 2cee3e9e6..b5ac8b12e 100644
--- a/DynamicData/Cache/Internal/OnItemRemoved.cs
+++ b/DynamicData/Cache/Internal/DisposeMany.cs
@@ -6,12 +6,12 @@
 
 namespace DynamicData.Cache.Internal
 {
-    internal class OnItemRemoved<TObject, TKey>
+    internal class DisposeMany<TObject, TKey>
     {
         private readonly IObservable<IChangeSet<TObject, TKey>> _source;
         private readonly Action<TObject> _removeAction;
 
-        public OnItemRemoved([NotNull] IObservable<IChangeSet<TObject, TKey>> source, Action<TObject> removeAction)
+        public DisposeMany([NotNull] IObservable<IChangeSet<TObject, TKey>> source, Action<TObject> removeAction)
         {
             _source = source ?? throw new ArgumentNullException(nameof(source));
             _removeAction = removeAction ?? throw new ArgumentNullException(nameof(removeAction));
diff --git a/DynamicData/Cache/ObservableCacheEx.cs b/DynamicData/Cache/ObservableCacheEx.cs
index 600e6e680..20cdd22e2 100644
--- a/DynamicData/Cache/ObservableCacheEx.cs
+++ b/DynamicData/Cache/ObservableCacheEx.cs
@@ -394,6 +394,24 @@ public static IObservable<IChangeSet<TObject, TKey>> SubscribeMany<TObject, TKey
             return new SubscribeMany<TObject, TKey>(source, subscriptionFactory).Run();
         }
 
+        
+        /// <summary>
+        /// Callback for each item as and when it is being added to the stream
+        /// </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="addAction">The add action.</param>
+        /// <returns></returns>
+        public static IObservable<IChangeSet<TObject, TKey>> OnItemAdded<TObject, TKey>(this IObservable<IChangeSet<TObject, TKey>> source, [NotNull] Action<TObject> addAction)
+        {
+            if (source == null) throw new ArgumentNullException(nameof(source));
+            if (addAction == null) throw new ArgumentNullException(nameof(addAction));
+
+            return source.Do(changes => changes.Where(c => c.Reason == ChangeReason.Add)
+                                               .ForEach(c => addAction(c.Current)));
+        }
+
         /// <summary>
         /// Callback for each item as and when it is being removed from the stream
         /// </summary>
@@ -412,7 +430,8 @@ public static IObservable<IChangeSet<TObject, TKey>> OnItemRemoved<TObject, TKey
             if (source == null) throw new ArgumentNullException(nameof(source));
             if (removeAction == null) throw new ArgumentNullException(nameof(removeAction));
 
-            return new OnItemRemoved<TObject, TKey>(source, removeAction).Run();
+            return source.Do(changes => changes.Where(c => c.Reason == ChangeReason.Remove)
+                                               .ForEach(c => removeAction(c.Current)));
         }
 
         /// <summary>
@@ -431,10 +450,7 @@ public static IObservable<IChangeSet<TObject, TKey>> OnItemUpdated<TObject, TKey
             if (updateAction == null) throw new ArgumentNullException(nameof(updateAction));
 
             return source.Do(changes => changes.Where(c => c.Reason == ChangeReason.Update)
-                .ForEach(c =>
-                {
-                    updateAction(c.Current, c.Previous.Value);
-                }));
+                                               .ForEach(c => updateAction(c.Current, c.Previous.Value)));
         }
 
         /// <summary>
@@ -452,11 +468,13 @@ public static IObservable<IChangeSet<TObject, TKey>> OnItemUpdated<TObject, TKey
         /// <exception cref="System.ArgumentNullException">source</exception>
         public static IObservable<IChangeSet<TObject, TKey>> DisposeMany<TObject, TKey>(this IObservable<IChangeSet<TObject, TKey>> source)
         {
-            return source.OnItemRemoved(t =>
-            {
-                var d = t as IDisposable;
-                d?.Dispose();
-            });
+            if (source == null) throw new ArgumentNullException(nameof(source));
+            return new DisposeMany<TObject, TKey>(source, t =>
+                {
+                    var d = t as IDisposable;
+                    d?.Dispose();
+                })
+                .Run();
         }
 
         /// <summary>