Merge pull request #231 from Didovgopoly/FixAndWhenSourceIsNotEmpty

Fix And operator
reactivemarbles · Jul 2, 2019 · d5347ab · d5347ab
2 parents f0d4ac1 + 58c479a
commit d5347ab
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 13 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/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/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/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;
             });
         }