Merge pull request #196 from Peter-B-/ExpireAfterBug

Expire after bug
reactivemarbles · Feb 11, 2019 · 9df4204 · 9df4204
2 parents 4b16e0a + 2efeb3e
commit 9df4204
Show file tree

Hide file tree

Showing 3 changed files with 65 additions and 2 deletions.
diff --git a/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs b/DynamicData.Tests/Cache/MergeManyWithKeyOverloadFixture.cs
@@ -25,6 +25,16 @@ public void InvokeObservable(bool value)
                 _changed.OnNext(value);
             }
 
+            public void CompleteObservable()
+            {
+                _changed.OnCompleted();
+            }
+
+            public void FailObservable(Exception ex)
+            {
+                _changed.OnError(ex);
+            }
+
             public IObservable<bool> Observable => _changed.AsObservable();
 
             public int Id { get; }
@@ -95,5 +105,51 @@ public void EverythingIsUnsubscribedWhenStreamIsDisposed()
             item.InvokeObservable(true);
             invoked.Should().BeFalse();
         }
+
+        [Fact]
+        public void SingleItemCompleteWillNotMergedStream()
+        {
+            var completed = false;
+            var stream =
+                _source.Connect()
+                    .MergeMany((o, key) => o.Observable)
+                    .Subscribe(
+                        _ => {},
+                        () => completed = true
+                        );
+
+            var item = new ObjectWithObservable(1);
+            _source.AddOrUpdate(item);
+
+            item.InvokeObservable(true);
+            item.CompleteObservable();
+
+            stream.Dispose();
+
+            completed.Should().BeFalse();
+        }
+
+        [Fact]
+        public void SingleItemFailWillNotFailMergedStream()
+        {
+            var failed = false;
+            var stream =
+                _source.Connect()
+                    .MergeMany((o, key) => o.Observable)
+                    .Subscribe(
+                        _ => { },
+                        ex => failed = true
+                    );
+
+            var item = new ObjectWithObservable(1);
+            _source.AddOrUpdate(item);
+
+            item.FailObservable(new Exception("Test exception"));
+
+            stream.Dispose();
+
+            failed.Should().BeFalse();
+        }
+
     }
 }
diff --git a/DynamicData.Tests/Cache/ObservableToObservableChangeSetFixture.cs b/DynamicData.Tests/Cache/ObservableToObservableChangeSetFixture.cs
@@ -177,7 +177,7 @@ public void ExpireAfterTimeDynamic()
             results.Data.Count.Should().Be(10, "Should be 10 items in the cache");
         }
 
-       // [Fact]
+        [Fact]
         public void ExpireAfterTimeDynamicWithKey()
         {
             var scheduler = new TestScheduler();

diff --git a/DynamicData/Cache/Internal/MergeMany.cs b/DynamicData/Cache/Internal/MergeMany.cs
@@ -26,7 +26,14 @@ public IObservable<TDestination> Run()
         {
             return Observable.Create<TDestination>
             (
-                observer => _source.SubscribeMany((t, key) => _observableSelector(t, key).Subscribe(observer))
+                observer => _source.SubscribeMany(
+                        (t, key) => _observableSelector(t, key)
+                            .Subscribe(
+                                observer.OnNext,
+                                ex => {},
+                                ( ) => {}
+                                )
+                        )
                     .Subscribe(t => { }, observer.OnError));
         }
     }