[Backports #6290] Akka:Streams Resolve `IAsyncEnumerator.DisposeAsync…

…` bug (#6296)

* Backports #6290

Fixes #6280

We no longer attempt to dispose IAsyncEnumerators inside Source stages due to the reasons outlined here: #6280 (comment)

This prevents the log from being filled with NotSupportedException warnings from failed DisposeAsync operations.

(cherry-picked from 3156272)

* post-merge fix

Co-authored-by: Aaron Stannard <aaron@petabridge.com>

src/core/Akka.Streams.Tests/Akka.Streams.Tests.csproj

@@ -5,6 +5,7 @@
     <AssemblyName>Akka.Streams.Tests</AssemblyName>
     <TargetFrameworks>$(NetFrameworkTestVersion);$(NetTestVersion);$(NetCoreTestVersion)</TargetFrameworks>
     <LangVersion>8.0</LangVersion>
+    <LangVersion>9</LangVersion>
   </PropertyGroup>
 
   <ItemGroup>

src/core/Akka.Streams.Tests/Dsl/AsyncEnumerableSpec.cs

@@ -25,7 +25,7 @@
 
 namespace Akka.Streams.Tests.Dsl
 {
-#if NETCOREAPP
+#if !NETFRAMEWORK // disabling this causes .NET Framework 4.7.2 builds to fail on Linux
     public class AsyncEnumerableSpec : AkkaSpec
     {
         private ActorMaterializer Materializer { get; }
@@ -227,6 +227,47 @@ public async Task AsyncEnumerableSource_Must_Cancel_Running_Source_If_Downstream
             }, Materializer);
         }
 
+        /// <summary>
+        /// Reproduction for https://github.com/akkadotnet/akka.net/issues/6280
+        /// </summary>
+        [Fact]
+        public async Task AsyncEnumerableSource_BugFix6280()
+        {
+            async IAsyncEnumerable<int> GenerateInts()
+            {
+                foreach (var i in Enumerable.Range(0, 100))
+                {
+                    if (i > 50)
+                        await Task.Delay(1000);
+                    yield return i;
+                }
+            }
+
+            var source = Source.From(GenerateInts);
+            var subscriber = this.CreateManualSubscriberProbe<int>();
+
+            await EventFilter.Warning().ExpectAsync(0, async () =>
+            {
+                var mat = source
+                    .WatchTermination(Keep.Right)
+                    .ToMaterialized(Sink.FromSubscriber(subscriber), Keep.Left);
+
+#pragma warning disable CS4014
+                var task = mat.Run(Materializer);
+#pragma warning restore CS4014
+
+                var subscription = await subscriber.ExpectSubscriptionAsync();
+                subscription.Request(50);
+                subscriber.ExpectNextN(Enumerable.Range(0, 50));
+                subscription.Request(10); // the iterator is going to start delaying 1000ms per item here
+                subscription.Cancel();
+
+
+                // The cancellation token inside the IAsyncEnumerable should be cancelled
+                await task;
+            });
+        }
+
         private static async IAsyncEnumerable<int> RangeAsync(int start, int count, 
             [EnumeratorCancellation] CancellationToken token = default)
         {
@@ -257,10 +298,7 @@ public async Task AsyncEnumerableSource_Must_Cancel_Running_Source_If_Downstream
         private static async IAsyncEnumerable<int> ProbeableRangeAsync(int start, int count, AtomicBoolean latch,
             [EnumeratorCancellation] CancellationToken token = default)
         {
-            token.Register(() =>
-            {
-                latch.GetAndSet(true);
-            });
+            token.Register(() => { latch.GetAndSet(true); });
             foreach (var i in Enumerable.Range(start, count))
             {
                 if(token.IsCancellationRequested)
@@ -269,8 +307,6 @@ public async Task AsyncEnumerableSource_Must_Cancel_Running_Source_If_Downstream
                 yield return i;
             }
         }
-
     }
-#else
 #endif
 }

src/core/Akka.Streams/Akka.Streams.csproj

@@ -7,7 +7,7 @@
         <TargetFrameworks>$(NetStandardLibVersion);$(NetLibVersion)</TargetFrameworks>
         <PackageTags>$(AkkaPackageTags);reactive;stream</PackageTags>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
-        <LangVersion>8.0</LangVersion>
+        <LangVersion>9</LangVersion>
     </PropertyGroup>
     <ItemGroup>
         <EmbeddedResource Include="reference.conf" />

src/core/Akka.Streams/Implementation/Fusing/Ops.cs

@@ -3871,20 +3871,8 @@ public override void OnDownstreamFinish(Exception cause)
             {
                 _completionCts.Cancel();
                 _completionCts.Dispose();
-
-                try
-                {
-                    _enumerator.DisposeAsync().GetAwaiter().GetResult();
-                }
-                catch (Exception ex)
-                {
-                    Log.Warning(ex, "Failed to dispose IAsyncEnumerator asynchronously");
-                }
-                finally
-                {
-                    CompleteStage();
-                    base.OnDownstreamFinish(cause);
-                }
+                CompleteStage();
+                base.OnDownstreamFinish(cause);
             }
 
         }