Merge branch 'master' of https://github.com/RolandPheasant/DynamicData

reactivemarbles · Jul 8, 2019 · 7e418d7 · 7e418d7
2 parents 4480643 + d5347ab
commit 7e418d7
Show file tree

Hide file tree

Showing 7 changed files with 88 additions and 15 deletions.
diff --git a/DynamicData.Tests/Cache/AndFixture.cs b/DynamicData.Tests/Cache/AndFixture.cs
@@ -95,5 +95,18 @@ public void UpdatingOneProducesOnlyOneUpdate()
             _results.Data.Count.Should().Be(1, "Cache should have no items");
             _results.Data.Items.First().Should().Be(personUpdated, "Should be updated person");
         }
+
+        [Fact]
+        public void StartingWithNonEmptySourceProducesNoResult()
+        {
+            var person = new Person("Adult", 50);
+            _source1.AddOrUpdate(person);
+
+            using (var result = CreateObservable().AsAggregator())
+            {
+                _results.Messages.Count.Should().Be(0, "Should have no updates");
+                result.Data.Count.Should().Be(0, "Cache should have no items");
+            }
+        }
     }
 }
diff --git a/DynamicData.Tests/Cache/FilterFixture.cs b/DynamicData.Tests/Cache/FilterFixture.cs
@@ -1,5 +1,6 @@
 using System;
 using System.Linq;
+using System.Reactive.Linq;
 using DynamicData.Tests.Domain;
 using FluentAssertions;
 using Xunit;
@@ -190,5 +191,24 @@ public void UpdateNotMatched()
             _results.Messages.Count.Should().Be(0, "Should be no updates");
             _results.Data.Count.Should().Be(0, "Should nothing cached");
         }
+
+        [Fact]
+        public void DuplicateKeyWithMerge()
+        {
+            const string key = "Adult1";
+            var newperson = new Person(key, 30);
+
+            using (var results = _source.Connect()
+                .Merge(_source.Connect())
+                .Filter(p => p.Age > 20).AsAggregator())
+            {
+                _source.AddOrUpdate(newperson); // previously this would throw an exception
+
+                results.Messages.Count.Should().Be(2, "Should be 2 messages");
+                results.Messages[0].Adds.Should().Be(1, "Should be 1 add");
+                results.Messages[1].Updates.Should().Be(1, "Should be 1 update");
+                results.Data.Count.Should().Be(1, "Should be cached");
+            }
+        }
     }
 }
diff --git a/DynamicData.Tests/List/AndFixture.cs b/DynamicData.Tests/List/AndFixture.cs
@@ -90,5 +90,16 @@ public void ClearOneClearsResult()
             _source1.Clear();
             _results.Data.Count.Should().Be(0);
         }
+
+        [Fact]
+        public void StartingWithNonEmptySourceProducesNoResult()
+        {
+            _source1.Add(1);
+
+            using (var result = CreateObservable().AsAggregator())
+            {
+                result.Data.Count.Should().Be(0);
+            }
+        }
     }
 }
diff --git a/DynamicData.Tests/List/OrFixture.cs b/DynamicData.Tests/List/OrFixture.cs
@@ -25,7 +25,27 @@ protected override IObservable<IChangeSet<int>> CreateObservable()
         }
     }
 
-
+    public class OrRefreshFixture
+    {
+        [Fact]
+        public void RefreshPassesThrough()
+        {
+            SourceList<Item> source1 = new SourceList<Item>();
+            source1.Add(new Item("A"));
+            SourceList<Item> source2 = new SourceList<Item>();
+            source2.Add(new Item("B"));
+
+            var list = new List<IObservable<IChangeSet<Item>>> { source1.Connect().AutoRefresh(), source2.Connect().AutoRefresh() };
+            var results = list.Or().AsAggregator();
+            source1.Items.ElementAt(0).Name = "Test";
+
+            results.Data.Count.Should().Be(2);
+            results.Messages.Count.Should().Be(3);
+            results.Messages[2].Refreshes.Should().Be(1);
+            results.Messages[2].First().Item.Current.Should().Be(source1.Items.First());
+        }
+    }
+
     public abstract class OrFixtureBase: IDisposable
     {
         protected ISourceList<int> _source1;

diff --git a/DynamicData/Cache/Internal/Combiner.cs b/DynamicData/Cache/Internal/Combiner.cs
@@ -30,15 +30,16 @@ public IDisposable Subscribe(IObservable<IChangeSet<TObject, TKey>>[] source)
             var disposable = new CompositeDisposable();
             lock (_locker)
             {
-                foreach (var item in source)
-                {
-                    var cache = new Cache<TObject, TKey>();
-                    _sourceCaches.Add(cache);
+                var caches = Enumerable.Range(0, source.Length).Select(_ => new Cache<TObject, TKey>());
+                _sourceCaches.AddRange(caches);
 
-                    var subsription = item.Subscribe(updates => Update(cache, updates));
-                    disposable.Add(subsription);
+                foreach (var pair in source.Zip(_sourceCaches, (item, cache) => new { Item = item, Cache = cache }))
+                {
+                    var subscription = pair.Item.Subscribe(updates => Update(pair.Cache, updates));
+                    disposable.Add(subscription);
                 }
             }
+
             return disposable;
         }
 

diff --git a/DynamicData/Cache/Internal/FilterEx.cs b/DynamicData/Cache/Internal/FilterEx.cs
@@ -46,7 +46,7 @@ public static void FilterChanges<TObject, TKey>(this ChangeAwareCache<TObject, T
                     {
                         var current = change.Current;
                         if (predicate(current))
-                            cache.Add(current, key);
+                            cache.AddOrUpdate(current, key);
                     }
                         break;
                     case ChangeReason.Update:

diff --git a/DynamicData/List/Internal/Combiner.cs b/DynamicData/List/Internal/Combiner.cs
@@ -25,26 +25,28 @@ public IObservable<IChangeSet<T>> Run()
             return Observable.Create<IChangeSet<T>>(observer =>
             {
                 var disposable = new CompositeDisposable();
-                var sourceLists = new List<ReferenceCountTracker<T>>();
+
                 var resultList = new ChangeAwareListWithRefCounts<T>();
 
                 lock (_locker)
                 {
-                    foreach (var item in _source)
-                    {
-                        var list = new ReferenceCountTracker<T>();
-                        sourceLists.Add(list);
+                    var sourceLists = Enumerable.Range(0, _source.Count)
+                        .Select(_ => new ReferenceCountTracker<T>())
+                        .ToList(); 
 
-                        disposable.Add(item.Synchronize(_locker).Subscribe(changes =>
+                    foreach (var pair in _source.Zip(sourceLists, (item, list) => new { Item = item, List = list }))
+                    {
+                        disposable.Add(pair.Item.Synchronize(_locker).Subscribe(changes =>
                         {
-                            CloneSourceList(list, changes);
+                            CloneSourceList(pair.List, changes);
 
                             var notifications = UpdateResultList(changes, sourceLists, resultList);
                             if (notifications.Count != 0)
                                 observer.OnNext(notifications);
                         }));
                     }
                 }
+
                 return disposable;
             });
         }
@@ -90,7 +92,13 @@ private IChangeSet<T> UpdateResultList(IChangeSet<T> changes, List<ReferenceCoun
                 if (shouldBeInResult)
                 {
                     if (!isInResult)
+                    {
                         resultList.Add(item);
+                    }
+                    else if (change.Reason == ListChangeReason.Refresh)
+                    {
+                        resultList.Refresh(change.Current);
+                    }
                 }
                 else
                 {