Making AsyncIterator a top-level class.

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

@@ -8,136 +8,133 @@ using System.Threading.Tasks;
 
 namespace System.Linq
 {
-    public static partial class AsyncEnumerable
+    internal abstract class AsyncIterator<TSource> : IAsyncEnumerable<TSource>, IAsyncEnumerator<TSource>
     {
-        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()
         {
-            private readonly int threadId;
+            threadId = Environment.CurrentManagedThreadId;
+        }
 
-            private CancellationTokenSource cancellationTokenSource;
-            private bool currentIsInvalid = true;
+        public IAsyncEnumerator<TSource> GetAsyncEnumerator()
+        {
+            var enumerator = state == AsyncIteratorState.New && threadId == Environment.CurrentManagedThreadId ?
+                this :
+                Clone();
 
-            internal TSource current;
-            internal AsyncIteratorState state = AsyncIteratorState.New;
+            enumerator.state = AsyncIteratorState.Allocated;
+            enumerator.cancellationTokenSource = new CancellationTokenSource();
 
-            protected AsyncIterator()
+            try
             {
-                threadId = Environment.CurrentManagedThreadId;
+                enumerator.OnGetEnumerator();
             }
-
-            public IAsyncEnumerator<TSource> GetAsyncEnumerator()
+            catch
             {
-                var enumerator = state == AsyncIteratorState.New && threadId == Environment.CurrentManagedThreadId ?
-                    this :
-                    Clone();
+                enumerator.DisposeAsync(); // REVIEW: fire-and-forget?
+                throw;
+            }
 
-                enumerator.state = AsyncIteratorState.Allocated;
-                enumerator.cancellationTokenSource = new CancellationTokenSource();
+            return enumerator;
+        }
 
-                try
-                {
-                    enumerator.OnGetEnumerator();
-                }
-                catch
+        public virtual Task DisposeAsync()
+        {
+            if (cancellationTokenSource != null)
+            {
+                if (!cancellationTokenSource.IsCancellationRequested)
                 {
-                    enumerator.DisposeAsync(); // REVIEW: fire-and-forget?
-                    throw;
+                    cancellationTokenSource.Cancel();
                 }
 
-                return enumerator;
+                cancellationTokenSource.Dispose();
             }
 
-            public virtual Task DisposeAsync()
-            {
-                if (cancellationTokenSource != null)
-                {
-                    if (!cancellationTokenSource.IsCancellationRequested)
-                    {
-                        cancellationTokenSource.Cancel();
-                    }
+            current = default(TSource);
+            state = AsyncIteratorState.Disposed;
 
-                    cancellationTokenSource.Dispose();
-                }
+            return TaskExt.CompletedTask;
+        }
 
-                current = default(TSource);
-                state = AsyncIteratorState.Disposed;
+        public TSource Current
+        {
+            get
+            {
+                if (currentIsInvalid)
+                    throw new InvalidOperationException("Enumerator is in an invalid state");
 
-                return TaskExt.CompletedTask;
+                return current;
             }
+        }
 
-            public TSource Current
-            {
-                get
-                {
-                    if (currentIsInvalid)
-                        throw new InvalidOperationException("Enumerator is in an invalid state");
+        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
 
-                    return current;
-                }
+            if (state == AsyncIteratorState.Disposed)
+            {
+                return false;
             }
 
-            public async Task<bool> MoveNextAsync()
+            try
             {
-                // 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;
-                }
+                var result = await MoveNextCore().ConfigureAwait(false);
 
-                try
-                {
-                    var result = await MoveNextCore().ConfigureAwait(false);
-
-                    currentIsInvalid = !result; // if move next is false, invalid otherwise valid
+                currentIsInvalid = !result; // if move next is false, invalid otherwise valid
 
-                    return result;
-                }
-                catch
-                {
-                    currentIsInvalid = true;
-                    await DisposeAsync().ConfigureAwait(false);
-                    throw;
-                }
+                return result;
             }
-
-            public abstract AsyncIterator<TSource> Clone();
-
-            public virtual IAsyncEnumerable<TResult> Select<TResult>(Func<TSource, TResult> selector)
+            catch
             {
-                return new SelectEnumerableAsyncIterator<TSource, TResult>(this, selector);
+                currentIsInvalid = true;
+                await DisposeAsync().ConfigureAwait(false);
+                throw;
             }
+        }
 
-            public virtual IAsyncEnumerable<TResult> Select<TResult>(Func<TSource, Task<TResult>> selector)
-            {
-                return new SelectEnumerableAsyncIteratorWithTask<TSource, TResult>(this, selector);
-            }
+        public abstract AsyncIterator<TSource> Clone();
 
-            public virtual IAsyncEnumerable<TSource> Where(Func<TSource, bool> predicate)
-            {
-                return new WhereEnumerableAsyncIterator<TSource>(this, predicate);
-            }
+        public virtual IAsyncEnumerable<TResult> Select<TResult>(Func<TSource, TResult> selector)
+        {
+            return new AsyncEnumerable.SelectEnumerableAsyncIterator<TSource, TResult>(this, selector);
+        }
 
-            public virtual IAsyncEnumerable<TSource> Where(Func<TSource, Task<bool>> predicate)
-            {
-                return new WhereEnumerableAsyncIteratorWithTask<TSource>(this, predicate);
-            }
+        public virtual IAsyncEnumerable<TResult> Select<TResult>(Func<TSource, Task<TResult>> selector)
+        {
+            return new AsyncEnumerable.SelectEnumerableAsyncIteratorWithTask<TSource, TResult>(this, selector);
+        }
 
-            protected abstract Task<bool> MoveNextCore();
+        public virtual IAsyncEnumerable<TSource> Where(Func<TSource, bool> predicate)
+        {
+            return new AsyncEnumerable.WhereEnumerableAsyncIterator<TSource>(this, predicate);
+        }
 
-            protected virtual void OnGetEnumerator()
-            {
-            }
+        public virtual IAsyncEnumerable<TSource> Where(Func<TSource, Task<bool>> predicate)
+        {
+            return new AsyncEnumerable.WhereEnumerableAsyncIteratorWithTask<TSource>(this, predicate);
         }
 
-        internal enum AsyncIteratorState
+        protected abstract Task<bool> MoveNextCore();
+
+        protected virtual void OnGetEnumerator()
         {
-            New = 0,
-            Allocated = 1,
-            Iterating = 2,
-            Disposed = -1
         }
     }
+
+    internal enum AsyncIteratorState
+    {
+        New = 0,
+        Allocated = 1,
+        Iterating = 2,
+        Disposed = -1
+    }
 }

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

@@ -8,7 +8,7 @@ using System.Threading.Tasks;
 
 namespace System.Linq
 {
-    internal abstract class OrderedAsyncEnumerable<TElement> : AsyncEnumerable.AsyncIterator<TElement>, IOrderedAsyncEnumerable<TElement>
+    internal abstract class OrderedAsyncEnumerable<TElement> : AsyncIterator<TElement>, IOrderedAsyncEnumerable<TElement>
     {
         internal IOrderedEnumerable<TElement> enumerable;
         internal IAsyncEnumerable<TElement> source;
@@ -49,7 +49,7 @@ namespace System.Linq
             this.parent = parent;
         }
 
-        public override AsyncEnumerable.AsyncIterator<TElement> Clone()
+        public override AsyncIterator<TElement> Clone()
         {
             return new OrderedAsyncEnumerable<TElement, TKey>(source, keySelector, comparer, descending, parent);
         }
@@ -75,15 +75,15 @@ namespace System.Linq
         {
             switch (state)
             {
-                case AsyncEnumerable.AsyncIteratorState.Allocated:
+                case AsyncIteratorState.Allocated:
 
                     await Initialize().ConfigureAwait(false);
 
                     enumerator = enumerable.GetEnumerator();
-                    state = AsyncEnumerable.AsyncIteratorState.Iterating;
-                    goto case AsyncEnumerable.AsyncIteratorState.Iterating;
+                    state = AsyncIteratorState.Iterating;
+                    goto case AsyncIteratorState.Iterating;
 
-                case AsyncEnumerable.AsyncIteratorState.Iterating:
+                case AsyncIteratorState.Iterating:
                     if (enumerator.MoveNext())
                     {
                         current = enumerator.Current;
@@ -136,7 +136,7 @@ namespace System.Linq
             this.parent = parent;
         }
 
-        public override AsyncEnumerable.AsyncIterator<TElement> Clone()
+        public override AsyncIterator<TElement> Clone()
         {
             return new OrderedAsyncEnumerableWithTask<TElement, TKey>(source, keySelector, comparer, descending, parent);
         }
@@ -162,15 +162,15 @@ namespace System.Linq
         {
             switch (state)
             {
-                case AsyncEnumerable.AsyncIteratorState.Allocated:
+                case AsyncIteratorState.Allocated:
 
                     await Initialize().ConfigureAwait(false);
 
                     enumerator = enumerable.GetEnumerator();
-                    state = AsyncEnumerable.AsyncIteratorState.Iterating;
-                    goto case AsyncEnumerable.AsyncIteratorState.Iterating;
+                    state = AsyncIteratorState.Iterating;
+                    goto case AsyncIteratorState.Iterating;
 
-                case AsyncEnumerable.AsyncIteratorState.Iterating:
+                case AsyncIteratorState.Iterating:
                     if (enumerator.MoveNext())
                     {
                         current = enumerator.Current;