Add property change notification tracking to ReactiveCollection

ReactiveXaml.Tests/ReactiveCollectionTest.cs

@@ -0,0 +1,140 @@
+using Antireptilia;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using System;
+using System.Linq;
+using System.Collections.Generic;
+using ReactiveXaml;
+using System.IO;
+using System.Text;
+using ReactiveXaml.Tests;
+
+namespace Antireptilia.Tests
+{
+    [TestClass()]
+    public class ReactiveCollectionTest : IEnableLogger
+    {
+        [TestMethod()]
+        [DeploymentItem("Antireptilia.exe")]
+        public void CollectionCountChangedTest()
+        {
+            var fixture = new ReactiveCollection<int>();
+            var output = new List<int>();
+            fixture.CollectionCountChanged.Subscribe(output.Add);
+
+            fixture.Add(10);
+            fixture.Add(20);
+            fixture.Add(30);
+            fixture.RemoveAt(1);
+            fixture.Clear();
+
+            var results = new[]{1,2,3,2,0};
+            Assert.AreEqual(results.Length, output.Count);
+            results.Zip(output, (expected, actual) => new {expected,actual})
+                   .Run(x => Assert.AreEqual(x.expected, x.actual));
+        }
+
+        [TestMethod()]
+        [DeploymentItem("Antireptilia.exe")]
+        public void ItemsAddedAndRemovedTest()
+        {
+            var fixture = new ReactiveCollection<int>();
+            var added = new List<int>();
+            var removed = new List<int>();
+            fixture.ItemsAdded.Subscribe(added.Add);
+            fixture.ItemsRemoved.Subscribe(removed.Add);
+
+            fixture.Add(10);
+            fixture.Add(20);
+            fixture.Add(30);
+            fixture.RemoveAt(1);
+            fixture.Clear();
+
+            var added_results = new[]{10,20,30};
+            Assert.AreEqual(added_results.Length, added.Count);
+            added_results.Zip(added, (expected, actual) => new {expected,actual})
+                         .Run(x => Assert.AreEqual(x.expected, x.actual));
+
+            var removed_results = new[]{20};
+            Assert.AreEqual(removed_results.Length, removed.Count);
+            removed_results.Zip(removed, (expected, actual) => new {expected,actual})
+                           .Run(x => Assert.AreEqual(x.expected, x.actual));
+        }
+
+        [TestMethod()]
+        [DeploymentItem("Antireptilia.exe")]
+        public void ReactiveCollectionIsRoundTrippable()
+        {
+            var output = new[] {"Foo", "Bar", "Baz", "Bamf"};
+            var fixture = new ReactiveCollection<string>(output);
+
+            string json = JSONHelper.Serialize(fixture);
+            var results = JSONHelper.Deserialize<ReactiveCollection<string>>(json);
+            this.Log().Debug(json);
+
+            output.Zip(results, (expected, actual) => new { expected, actual })
+                  .Run(x => Assert.AreEqual(x.expected, x.actual));
+
+            bool should_die = true;
+            results.ItemsAdded.Subscribe(_ => should_die = false);
+            results.Add("Foobar");
+            Assert.IsFalse(should_die);
+        }
+
+        [TestMethod()]
+        [DeploymentItem("Antireptilia.exe")]
+        public void ChangeTrackingShouldFireNotifications()
+        {
+            var fixture = new ReactiveCollection<TestFixture>() { ChangeTrackingEnabled = true };
+            var output = new List<Tuple<TestFixture, string>>();
+            var item1 = new TestFixture() { IsOnlyOneWord = "Foo" };
+            var item2 = new TestFixture() { IsOnlyOneWord = "Bar" };
+
+            fixture.ItemPropertyChanged.Subscribe(x => {
+                output.Add(new Tuple<TestFixture,string>((TestFixture)x.Sender, x.PropertyName));
+            });
+
+            fixture.Add(item1);
+            fixture.Add(item2);
+
+            item1.IsOnlyOneWord = "Baz";
+            Assert.AreEqual(1, output.Count);
+            item2.IsNotNullString = "FooBar";
+            Assert.AreEqual(2, output.Count);
+
+            fixture.Remove(item2);
+            item2.IsNotNullString = "FooBarBaz";
+            Assert.AreEqual(2, output.Count);
+
+            fixture.ChangeTrackingEnabled = false;
+            item1.IsNotNullString = "Bamf";
+            Assert.AreEqual(2, output.Count);
+
+            new[]{item1, item2}.Zip(output.Select(x => x.Item1), (expected, actual) => new { expected, actual })
+                .Run(x => Assert.AreEqual(x.expected, x.actual));
+            new[]{"IsOnlyOneWord", "IsNotNullString"}.Zip(output.Select(x => x.Item2), (expected, actual) => new { expected, actual })
+                .Run(x => Assert.AreEqual(x.expected, x.actual));
+        }
+    }
+
+    public class JSONHelper
+    {
+        public static string Serialize<T>(T obj)
+        {
+            var serializer = new System.Runtime.Serialization.Json.DataContractJsonSerializer(obj.GetType());
+            var ms = new MemoryStream();
+            serializer.WriteObject(ms, obj);
+            string retVal = Encoding.Default.GetString(ms.ToArray());
+            return retVal;
+        }
+
+        public static T Deserialize<T>(string json)
+        {
+            var obj = Activator.CreateInstance<T>();
+            var ms = new MemoryStream(Encoding.Unicode.GetBytes(json));
+            var serializer = new System.Runtime.Serialization.Json.DataContractJsonSerializer(obj.GetType());
+            obj = (T)serializer.ReadObject(ms);
+            ms.Close();
+            return obj;
+        }
+    }
+}

ReactiveXaml.Tests/ReactiveXaml.Tests.csproj

@@ -60,6 +60,7 @@
     <Reference Include="System.Reactive">
       <HintPath>..\ext\System.Reactive.dll</HintPath>
     </Reference>
+    <Reference Include="System.Runtime.Serialization" />
     <Reference Include="System.Xml" />
     <Reference Include="System.Xml.Linq" />
     <Reference Include="WindowsBase" />
@@ -73,6 +74,7 @@
     <Compile Include="ObservableAsPropertyHelperTest.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="QueuedAsyncMRUCacheTest.cs" />
+    <Compile Include="ReactiveCollectionTest.cs" />
     <Compile Include="ReactiveCommandTest.cs" />
     <Compile Include="ReactiveObjectTest.cs" />
     <Compile Include="ReactiveValidatedObjectTest.cs" />

ReactiveXaml/Interfaces.cs

@@ -12,7 +12,7 @@ namespace ReactiveXaml
     public class ObservedChange<TSender, TValue>
     {
         public TSender Sender { get; set; }
-        public TValue Value { get; set; }
+        public string PropertyName { get; set; }
     }
 
     public interface IReactiveNotifyPropertyChanged : INotifyPropertyChanged, IObservable<PropertyChangedEventArgs> { }
@@ -22,13 +22,21 @@ namespace ReactiveXaml
         public static IObservable<ObservedChange<TSender, TValue>> ObservableForProperty<TSender, TValue>(this TSender This, string propertyName)
             where TSender : IReactiveNotifyPropertyChanged
         {
-            var prop = This.GetType().GetProperty(propertyName);
-
             return This.Where(x => x.PropertyName == propertyName)
-                       .Select(x => new ObservedChange<TSender, TValue> { Sender = This, Value = (TValue)prop.GetValue(This, null) });
+                       .Select(x => new ObservedChange<TSender, TValue> { Sender = This, PropertyName = x.PropertyName });
         }
     }
 
+    public interface IReactiveCollection<T> : IEnumerable<T>, IList<T>, INotifyCollectionChanged
+    {
+        IObservable<T> ItemsAdded { get; }
+        IObservable<T> ItemsRemoved { get; }
+        IObservable<int> CollectionCountChanged { get; }
+
+        bool ChangeTrackingEnabled { get; set; }
+        IObservable<ObservedChange<T, object>> ItemPropertyChanged { get; }
+    }
+
     public interface IReactiveCommand : ICommand, IObservable<object> 
     {
         IObservable<bool> CanExecuteObservable {get;} 

ReactiveXaml/ReactiveCollection.cs

@@ -0,0 +1,172 @@
+using System;
+using System.Collections;
+using System.Collections.Generic;
+using System.Collections.ObjectModel;
+using System.ComponentModel;
+using System.Collections.Specialized;
+using System.Linq;
+using System.Runtime.Serialization;
+
+namespace ReactiveXaml
+{
+    [Serializable]
+    public class ReactiveCollection<T> : ObservableCollection<T>, IReactiveCollection<T>, INotifyPropertyChanged, IDisposable
+    {
+        public ReactiveCollection() { setupRx(); }
+        public ReactiveCollection(IEnumerable<T> List) : base(List) { setupRx(); }
+
+        [OnDeserialized]
+        void setupRx(StreamingContext _) { setupRx(); }
+
+        void setupRx()
+        {
+            var coll_changed = Observable.FromEvent<NotifyCollectionChangedEventArgs>(this, "CollectionChanged");
+
+            coll_changed.Subscribe(x => {
+                int a = 1;
+
+                Console.WriteLine(x.EventArgs.Action);
+                Console.WriteLine(x.EventArgs.OldItems);
+            });
+
+            ItemsAdded = coll_changed
+                .Where(x => x.EventArgs.Action == NotifyCollectionChangedAction.Add || x.EventArgs.Action == NotifyCollectionChangedAction.Replace)
+                .SelectMany(x => (x.EventArgs.NewItems != null ? x.EventArgs.NewItems.OfType<T>() : Enumerable.Empty<T>()).ToObservable());
+
+            ItemsRemoved = coll_changed
+                .Where(x => x.EventArgs.Action == NotifyCollectionChangedAction.Remove || x.EventArgs.Action == NotifyCollectionChangedAction.Replace || x.EventArgs.Action == NotifyCollectionChangedAction.Reset)
+                .SelectMany(x => (x.EventArgs.OldItems != null ? x.EventArgs.OldItems.OfType<T>() : Enumerable.Empty<T>()).ToObservable());
+
+            CollectionCountChanged = coll_changed
+                .Select(x => this.Count)
+                .DistinctUntilChanged();
+
+            _ItemPropertyChanged = new Subject<ObservedChange<T,object>>();
+
+            ItemsAdded.Subscribe(x => {
+                if (propertyChangeWatchers == null)
+                    return;
+                var item = x as IReactiveNotifyPropertyChanged;
+                if (item == null)
+                    return;
+                propertyChangeWatchers.Add(x, item.Subscribe(change => 
+                    _ItemPropertyChanged.OnNext(new ObservedChange<T,object>() { Sender = x, PropertyName = change.PropertyName })));
+            });
+
+            ItemsRemoved.Subscribe(x => {
+                if (propertyChangeWatchers == null)
+                    return;
+                if (propertyChangeWatchers.ContainsKey(x)) {
+                    propertyChangeWatchers[x].Dispose();
+                    propertyChangeWatchers.Remove(x);
+                }
+            });
+        }
+
+        [NonSerialized]
+        IObservable<T> _ItemsAdded;
+        public IObservable<T> ItemsAdded {
+            get { return _ItemsAdded; }
+            protected set { _ItemsAdded = value; }
+        }
+
+        [NonSerialized]
+        IObservable<T> _ItemsRemoved;
+        public IObservable<T> ItemsRemoved {
+            get { return _ItemsRemoved; }
+            set { _ItemsRemoved = value; }
+        }
+
+        [NonSerialized]
+        IObservable<int> _CollectionCountChanged;
+        public IObservable<int> CollectionCountChanged { 
+            get { return _CollectionCountChanged; }
+            set { _CollectionCountChanged = value; }
+        }
+
+        [NonSerialized]
+        Subject<ObservedChange<T, object>> _ItemPropertyChanged;
+        public IObservable<ObservedChange<T, object>> ItemPropertyChanged {
+            get { return _ItemPropertyChanged; }
+        }
+
+        public bool ChangeTrackingEnabled {
+            get { return (propertyChangeWatchers != null); }
+            set {
+                if ((propertyChangeWatchers != null) == value)
+                    return;
+                if (propertyChangeWatchers == null) {
+                    propertyChangeWatchers = new Dictionary<object,IDisposable>();
+                } else {
+                    releasePropChangeWatchers();
+                    propertyChangeWatchers = null;
+                }
+            }
+        }
+
+        [NonSerialized]
+        Dictionary<object, IDisposable> _propertyChangeWatchers;
+        Dictionary<object, IDisposable> propertyChangeWatchers {
+            get { return _propertyChangeWatchers; }
+            set { _propertyChangeWatchers = value; }
+        }
+
+        protected void releasePropChangeWatchers()
+        {
+            if (propertyChangeWatchers == null) {
+                return;
+            }
+            propertyChangeWatchers.Values.Run(x => x.Dispose());
+            propertyChangeWatchers.Clear();
+        }
+
+        protected override void ClearItems()
+        {
+            // N.B: Reset doesn't give us the items that were cleared out,
+            // we have to release the watchers or else we leak them.
+            releasePropChangeWatchers();
+            base.ClearItems();
+        }
+
+        public void Dispose()
+        {
+            ChangeTrackingEnabled = false;
+        }
+
+        [field: NonSerialized]
+        public override event NotifyCollectionChangedEventHandler CollectionChanged;
+
+        [field: NonSerialized]
+        private PropertyChangedEventHandler _propertyChangedEventHandler;
+
+        event PropertyChangedEventHandler INotifyPropertyChanged.PropertyChanged {
+            add {
+                _propertyChangedEventHandler = Delegate.Combine(_propertyChangedEventHandler, value) as PropertyChangedEventHandler;
+            }
+            remove {
+                _propertyChangedEventHandler = Delegate.Remove(_propertyChangedEventHandler, value) as PropertyChangedEventHandler;
+            }
+        }
+
+        protected override void OnCollectionChanged(NotifyCollectionChangedEventArgs e)
+        {
+            NotifyCollectionChangedEventHandler handler = CollectionChanged;
+
+            if (handler != null) {
+                handler(this, e);
+            }
+        }
+
+        protected override void OnPropertyChanged(PropertyChangedEventArgs e)
+        {
+            PropertyChangedEventHandler handler = _propertyChangedEventHandler;
+
+            if (handler != null) {
+                handler(this, e);
+            }
+        }
+
+    }
+}
+
+// vim: tw=120 ts=4 sw=4 et enc=utf8 :

ReactiveXaml/ReactiveXaml.csproj

@@ -70,6 +70,7 @@
     <Compile Include="MemoizingMRUCache.cs" />
     <Compile Include="ObservableAsPropertyHelper.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
+    <Compile Include="ReactiveCollection.cs" />
     <Compile Include="ReactiveCommand.cs" />
     <Compile Include="ReactiveObject.cs" />
     <Compile Include="Schedulers.cs" />