Fix OnErrorResumeNext not disposing the old source.

Author: akarnokd

async-enumerable-dotnet-test/OnErrorResumeNextTest.cs

@@ -33,5 +33,16 @@ await AsyncEnumerable.Error<int>(new InvalidOperationException())
                 .OnErrorResumeNext(e => throw e)
                 .AssertFailure(typeof(AggregateException));
         }
+
+        [Fact]
+        public async void Error_DisposeSource()
+        {
+            var disposed = false;
+            await AsyncEnumerable.Error<int>(new Exception())
+                .DoOnDispose(() => disposed = true)
+                .OnErrorResumeNext(e => AsyncEnumerable.Empty<int>())
+                .AssertResult();
+            Assert.True(disposed);
+        }
     }
 }

async-enumerable-dotnet/impl/OnErrorResumeNext.cs

@@ -46,7 +46,12 @@ public OnErrorResumeNextEnumerator(IAsyncEnumerator<T> source, Func<Exception, I
 
             public ValueTask DisposeAsync()
             {
-                return _source.DisposeAsync();
+                var src = _source;
+                if (src != null)
+                {
+                    return src.DisposeAsync();
+                }
+                return new ValueTask();
             }
 
             public async ValueTask<bool> MoveNextAsync()
@@ -62,10 +67,18 @@ public async ValueTask<bool> MoveNextAsync()
                 }
                 catch (Exception ex)
                 {
+                    var old = _source;
+                    _source = null;
+                    if (old != null)
+                    {
+                        await old.DisposeAsync();
+                    }
+
                     IAsyncEnumerator<T> en;
 
                     try
                     {
+
                         en = _handler(ex).GetAsyncEnumerator(_ct);
                     }
                     catch (Exception exc)