Use async iterators for OfType.

Author: bartdesmet

Ix.NET/Source/System.Linq.Async/System/Linq/Operators/OfType.cs

@@ -3,20 +3,45 @@
 // 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
 {
     public static partial class AsyncEnumerable
     {
+        // REVIEW: This is a non-standard LINQ operator, because we don't have a non-generic IAsyncEnumerable.
+        //
+        //         Unfortunately, this has limited use because it requires the source to be IAsyncEnumerable<object>,
+        //         thus it doesn't bind for value types. Adding a first generic parameter for the element type of
+        //         the source is not an option, because it would require users to specify two type arguments, unlike
+        //         what's done in Enumerable.OfType. Should we move this method to Ix, thus doing away with OfType
+        //         in the API surface altogether?
+
         public static IAsyncEnumerable<TResult> OfType<TResult>(this IAsyncEnumerable<object> source)
         {
             if (source == null)
                 throw Error.ArgumentNull(nameof(source));
 
+#if USE_ASYNC_ITERATOR
+            return Create(Core);
+
+            async IAsyncEnumerator<TResult> Core(CancellationToken cancellationToken)
+            {
+                await foreach (var obj in AsyncEnumerableExtensions.WithCancellation(source, cancellationToken).ConfigureAwait(false))
+                {
+                    if (obj is TResult result)
+                    {
+                        yield return result;
+                    }
+                }
+            }
+#else
             return new OfTypeAsyncIterator<TResult>(source);
+#endif
         }
 
+#if !USE_ASYNC_ITERATOR
         private sealed class OfTypeAsyncIterator<TResult> : AsyncIterator<TResult>
         {
             private readonly IAsyncEnumerable<object> _source;
@@ -70,5 +95,6 @@ protected override async ValueTask<bool> MoveNextCore()
                 return false;
             }
         }
+#endif
     }
 }