Moving Standard Query Operators to System.Linq.Async.

Ix.NET/Source/System.Interactive.Async.Tests/AsyncTests.Exceptions.cs

@@ -536,10 +536,10 @@ namespace Tests
         [Fact]
         public void Retry_Null()
         {
-            AssertThrows<ArgumentNullException>(() => AsyncEnumerable.Retry<int>(default(IAsyncEnumerable<int>)));
+            AssertThrows<ArgumentNullException>(() => AsyncEnumerableEx.Retry<int>(default(IAsyncEnumerable<int>)));
 
-            AssertThrows<ArgumentNullException>(() => AsyncEnumerable.Retry<int>(default(IAsyncEnumerable<int>), 1));
-            AssertThrows<ArgumentOutOfRangeException>(() => AsyncEnumerable.Retry<int>(AsyncEnumerable.Return(42), -1));
+            AssertThrows<ArgumentNullException>(() => AsyncEnumerableEx.Retry<int>(default(IAsyncEnumerable<int>), 1));
+            AssertThrows<ArgumentOutOfRangeException>(() => AsyncEnumerableEx.Retry<int>(AsyncEnumerable.Return(42), -1));
         }
 
         [Fact]

Ix.NET/Source/System.Interactive.Async/AsyncIterator.cs

@@ -103,26 +103,6 @@ namespace System.Linq
 
         public abstract AsyncIterator<TSource> Clone();
 
-        public virtual IAsyncEnumerable<TResult> Select<TResult>(Func<TSource, TResult> selector)
-        {
-            return new AsyncEnumerable.SelectEnumerableAsyncIterator<TSource, TResult>(this, selector);
-        }
-
-        public virtual IAsyncEnumerable<TResult> Select<TResult>(Func<TSource, Task<TResult>> selector)
-        {
-            return new AsyncEnumerable.SelectEnumerableAsyncIteratorWithTask<TSource, TResult>(this, selector);
-        }
-
-        public virtual IAsyncEnumerable<TSource> Where(Func<TSource, bool> predicate)
-        {
-            return new AsyncEnumerable.WhereEnumerableAsyncIterator<TSource>(this, predicate);
-        }
-
-        public virtual IAsyncEnumerable<TSource> Where(Func<TSource, Task<bool>> predicate)
-        {
-            return new AsyncEnumerable.WhereEnumerableAsyncIteratorWithTask<TSource>(this, predicate);
-        }
-
         protected abstract Task<bool> MoveNextCore();
 
         protected virtual void OnGetEnumerator()

Ix.NET/Source/System.Interactive.Async/System.Interactive.Async.csproj

@@ -11,63 +11,12 @@
     <EmbeddedResource Include="Properties\System.Interactive.Async.rd.xml" />
   </ItemGroup>
 
-  <ItemGroup>
-    <None Include="Average.Generated.cs">
-      <DesignTime>True</DesignTime>
-      <AutoGen>True</AutoGen>
-      <DependentUpon>Average.Generated.tt</DependentUpon>
-    </None>
-    <None Include="MinMax.Generated.cs">
-      <DesignTime>True</DesignTime>
-      <AutoGen>True</AutoGen>
-      <DependentUpon>MinMax.Generated.tt</DependentUpon>
-    </None>
-    <None Include="Sum.Generated.cs">
-      <DesignTime>True</DesignTime>
-      <AutoGen>True</AutoGen>
-      <DependentUpon>Sum.Generated.tt</DependentUpon>
-    </None>
-  </ItemGroup>
-
   <ItemGroup>
     <ProjectReference Include="..\System.Linq.Async\System.Linq.Async.csproj" />
   </ItemGroup>
 
-  <ItemGroup>
-    <None Update="Average.Generated.tt">
-      <Generator>TextTemplatingFileGenerator</Generator>
-      <LastGenOutput>Average.Generated.cs</LastGenOutput>
-    </None>
-    <None Update="MinMax.Generated.tt">
-      <Generator>TextTemplatingFileGenerator</Generator>
-      <LastGenOutput>MinMax.Generated.cs</LastGenOutput>
-    </None>
-    <None Update="Sum.Generated.tt">
-      <Generator>TextTemplatingFileGenerator</Generator>
-      <LastGenOutput>Sum.Generated.cs</LastGenOutput>
-    </None>
-  </ItemGroup>
-
   <ItemGroup>
     <Service Include="{508349b6-6b84-4df5-91f0-309beebad82d}" />
   </ItemGroup>
 
-  <ItemGroup>
-    <Compile Update="Average.Generated.cs">
-      <DesignTime>True</DesignTime>
-      <AutoGen>True</AutoGen>
-      <DependentUpon>Average.Generated.tt</DependentUpon>
-    </Compile>
-    <Compile Update="MinMax.Generated.cs">
-      <DesignTime>True</DesignTime>
-      <AutoGen>True</AutoGen>
-      <DependentUpon>MinMax.Generated.tt</DependentUpon>
-    </Compile>
-    <Compile Update="Sum.Generated.cs">
-      <DesignTime>True</DesignTime>
-      <AutoGen>True</AutoGen>
-      <DependentUpon>Sum.Generated.tt</DependentUpon>
-    </Compile>
-  </ItemGroup>
-
 </Project>

Ix.NET/Source/System.Interactive.Async/Retry.cs → Ix.NET/Source/System.Interactive.Async/System/Linq/Operators/Retry.cs

@@ -6,7 +6,7 @@ using System.Collections.Generic;
 
 namespace System.Linq
 {
-    public static partial class AsyncEnumerable
+    public static partial class AsyncEnumerableEx
     {
         public static IAsyncEnumerable<TSource> Retry<TSource>(this IAsyncEnumerable<TSource> source)
         {

Ix.NET/Source/System.Linq.Async/System.Linq.Async.csproj

@@ -5,7 +5,40 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <Folder Include="System\Linq\Operators\" />
+    <Compile Update="System\Linq\Operators\Average.Generated.cs">
+      <DependentUpon>Average.Generated.tt</DependentUpon>
+      <DesignTime>True</DesignTime>
+      <AutoGen>True</AutoGen>
+    </Compile>
+    <Compile Update="System\Linq\Operators\MinMax.Generated.cs">
+      <DependentUpon>MinMax.Generated.tt</DependentUpon>
+      <DesignTime>True</DesignTime>
+      <AutoGen>True</AutoGen>
+    </Compile>
+    <Compile Update="System\Linq\Operators\Sum.Generated.cs">
+      <DependentUpon>Sum.Generated.tt</DependentUpon>
+      <DesignTime>True</DesignTime>
+      <AutoGen>True</AutoGen>
+    </Compile>
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="System\Linq\Operators\Average.Generated.tt">
+      <LastGenOutput>Average.Generated.cs</LastGenOutput>
+      <Generator>TextTemplatingFileGenerator</Generator>
+    </None>
+    <None Update="System\Linq\Operators\MinMax.Generated.tt">
+      <LastGenOutput>MinMax.Generated.cs</LastGenOutput>
+      <Generator>TextTemplatingFileGenerator</Generator>
+    </None>
+    <None Update="System\Linq\Operators\Sum.Generated.tt">
+      <LastGenOutput>Sum.Generated.cs</LastGenOutput>
+      <Generator>TextTemplatingFileGenerator</Generator>
+    </None>
+  </ItemGroup>
+
+  <ItemGroup>
+    <Service Include="{508349b6-6b84-4df5-91f0-309beebad82d}" />
   </ItemGroup>
 
 </Project>

Ix.NET/Source/System.Linq.Async/System/Linq/AsyncIterator.cs

@@ -0,0 +1,140 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information. 
+
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace System.Linq
+{
+    internal abstract class AsyncIterator<TSource> : IAsyncEnumerable<TSource>, IAsyncEnumerator<TSource>
+    {
+        private readonly int threadId;
+
+        private CancellationTokenSource cancellationTokenSource;
+        private bool currentIsInvalid = true;
+
+        internal TSource current;
+        internal AsyncIteratorState state = AsyncIteratorState.New;
+
+        protected AsyncIterator()
+        {
+            threadId = Environment.CurrentManagedThreadId;
+        }
+
+        public IAsyncEnumerator<TSource> GetAsyncEnumerator()
+        {
+            var enumerator = state == AsyncIteratorState.New && threadId == Environment.CurrentManagedThreadId ?
+                this :
+                Clone();
+
+            enumerator.state = AsyncIteratorState.Allocated;
+            enumerator.cancellationTokenSource = new CancellationTokenSource();
+
+            try
+            {
+                enumerator.OnGetEnumerator();
+            }
+            catch
+            {
+                enumerator.DisposeAsync(); // REVIEW: fire-and-forget?
+                throw;
+            }
+
+            return enumerator;
+        }
+
+        public virtual Task DisposeAsync()
+        {
+            if (cancellationTokenSource != null)
+            {
+                if (!cancellationTokenSource.IsCancellationRequested)
+                {
+                    cancellationTokenSource.Cancel();
+                }
+
+                cancellationTokenSource.Dispose();
+            }
+
+            current = default(TSource);
+            state = AsyncIteratorState.Disposed;
+
+            return TaskExt.CompletedTask;
+        }
+
+        public TSource Current
+        {
+            get
+            {
+                if (currentIsInvalid)
+                    throw new InvalidOperationException("Enumerator is in an invalid state");
+
+                return current;
+            }
+        }
+
+        public async Task<bool> MoveNextAsync()
+        {
+            // Note: MoveNext *must* be implemented as an async method to ensure
+            // that any exceptions thrown from the MoveNextCore call are handled 
+            // by the try/catch, whether they're sync or async
+
+            if (state == AsyncIteratorState.Disposed)
+            {
+                return false;
+            }
+
+            try
+            {
+                var result = await MoveNextCore().ConfigureAwait(false);
+
+                currentIsInvalid = !result; // if move next is false, invalid otherwise valid
+
+                return result;
+            }
+            catch
+            {
+                currentIsInvalid = true;
+                await DisposeAsync().ConfigureAwait(false);
+                throw;
+            }
+        }
+
+        public abstract AsyncIterator<TSource> Clone();
+
+        public virtual IAsyncEnumerable<TResult> Select<TResult>(Func<TSource, TResult> selector)
+        {
+            return new AsyncEnumerable.SelectEnumerableAsyncIterator<TSource, TResult>(this, selector);
+        }
+
+        public virtual IAsyncEnumerable<TResult> Select<TResult>(Func<TSource, Task<TResult>> selector)
+        {
+            return new AsyncEnumerable.SelectEnumerableAsyncIteratorWithTask<TSource, TResult>(this, selector);
+        }
+
+        public virtual IAsyncEnumerable<TSource> Where(Func<TSource, bool> predicate)
+        {
+            return new AsyncEnumerable.WhereEnumerableAsyncIterator<TSource>(this, predicate);
+        }
+
+        public virtual IAsyncEnumerable<TSource> Where(Func<TSource, Task<bool>> predicate)
+        {
+            return new AsyncEnumerable.WhereEnumerableAsyncIteratorWithTask<TSource>(this, predicate);
+        }
+
+        protected abstract Task<bool> MoveNextCore();
+
+        protected virtual void OnGetEnumerator()
+        {
+        }
+    }
+
+    internal enum AsyncIteratorState
+    {
+        New = 0,
+        Allocated = 1,
+        Iterating = 2,
+        Disposed = -1
+    }
+}

Ix.NET/Source/System.Linq.Async/System/Linq/EmptyArray.cs

@@ -0,0 +1,15 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information. 
+
+#if NO_ARRAY_EMPTY
+
+namespace System.Linq
+{
+    internal sealed class EmptyArray<TElement>
+    {
+        public static readonly TElement[] Value = new TElement[0];
+    }
+}
+
+#endif

Ix.NET/Source/System.Interactive.Async/Select.cs → Ix.NET/Source/System.Linq.Async/System/Linq/Operators/Select.cs

@@ -444,4 +444,4 @@ namespace System.Linq
             }
         }
     }
-}
+}

Ix.NET/Source/System.Linq.Async/System/Threading/Tasks/TaskExt.cs

@@ -0,0 +1,13 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information. 
+
+namespace System.Threading.Tasks
+{
+    internal static class TaskExt
+    {
+        public static readonly Task<bool> True = Task.FromResult(true);
+        public static readonly Task<bool> False = Task.FromResult(false);
+        public static readonly Task CompletedTask = True;
+    }
+}