Merge pull request #816 from akarnokd/FixCreateTaskInvalidCompletion

4.x: Fix accidental behavior change with Task-based Create methods completing when the body ends

Author: danielcweber

Rx.NET/Source/src/System.Reactive/Linq/QueryLanguage.Creation.cs

@@ -155,7 +155,6 @@ public void Dispose()
 
                     public void OnCompleted()
                     {
-                        _observer.OnCompleted();
                     }
 
                     public void OnError(Exception error)
@@ -233,7 +232,6 @@ public void Dispose()
 
                     public void OnCompleted()
                     {
-                        _observer.OnCompleted();
                     }
 
                     public void OnError(Exception error)

Rx.NET/Source/tests/Tests.System.Reactive/Tests/Linq/Observable/CreateAsyncTest.cs

@@ -705,5 +705,46 @@ public void CreateAsync_Action_Token()
             Assert.True(lst.Take(10).SequenceEqual(Enumerable.Repeat(42, 10)));
         }
 
+
+        [Fact]
+        public void CreateWithTaskDisposable_NoPrematureTermination()
+        {
+            var obs = Observable.Create<int>(async o =>
+            {
+                // avoid warning on async o due to no await
+                await Task.CompletedTask;
+
+                var inner = Observable.Range(1, 3);
+
+                return inner.Subscribe(x =>
+                {
+                    o.OnNext(x);
+                });
+            });
+
+            var result = obs.Take(1).Wait();
+        }
+
+        [Fact]
+        public void CreateWithTaskAction_NoPrematureTermination()
+        {
+            var obs = Observable.Create<int>(async o =>
+            {
+                // avoid warning on async o due to no await
+                await Task.CompletedTask;
+
+                var inner = Observable.Range(1, 3);
+
+                var d = inner.Subscribe(x =>
+                {
+                    o.OnNext(x);
+                });
+
+                Action a = () => d.Dispose();
+                return a;
+            });
+
+            var result = obs.Take(1).Wait();
+        }
     }
 }