Optimize Single.

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

@@ -102,14 +102,68 @@ namespace System.Linq
             }
         }
 
-        private static Task<TSource> SingleCore<TSource>(IAsyncEnumerable<TSource> source, Func<TSource, bool> predicate, CancellationToken cancellationToken)
+        private static async Task<TSource> SingleCore<TSource>(IAsyncEnumerable<TSource> source, Func<TSource, bool> predicate, CancellationToken cancellationToken)
         {
-            return source.Where(predicate).Single(cancellationToken);
+            var e = source.GetAsyncEnumerator(cancellationToken);
+
+            try
+            {
+                while (await e.MoveNextAsync().ConfigureAwait(false))
+                {
+                    var result = e.Current;
+
+                    if (predicate(result))
+                    {
+                        while (await e.MoveNextAsync().ConfigureAwait(false))
+                        {
+                            if (predicate(e.Current))
+                            {
+                                throw new InvalidOperationException(Strings.MORE_THAN_ONE_ELEMENT);
+                            }
+                        }
+
+                        return result;
+                    }
+                }
+
+                throw new InvalidOperationException(Strings.NO_ELEMENTS);
+            }
+            finally
+            {
+                await e.DisposeAsync().ConfigureAwait(false);
+            }
         }
 
-        private static Task<TSource> SingleCore<TSource>(IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate, CancellationToken cancellationToken)
+        private static async Task<TSource> SingleCore<TSource>(IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate, CancellationToken cancellationToken)
         {
-            return source.Where(predicate).Single(cancellationToken);
+            var e = source.GetAsyncEnumerator(cancellationToken);
+
+            try
+            {
+                while (await e.MoveNextAsync().ConfigureAwait(false))
+                {
+                    var result = e.Current;
+
+                    if (await predicate(result).ConfigureAwait(false))
+                    {
+                        while (await e.MoveNextAsync().ConfigureAwait(false))
+                        {
+                            if (await predicate(e.Current).ConfigureAwait(false))
+                            {
+                                throw new InvalidOperationException(Strings.MORE_THAN_ONE_ELEMENT);
+                            }
+                        }
+
+                        return result;
+                    }
+                }
+
+                throw new InvalidOperationException(Strings.NO_ELEMENTS);
+            }
+            finally
+            {
+                await e.DisposeAsync().ConfigureAwait(false);
+            }
         }
     }
 }

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

@@ -89,6 +89,7 @@ namespace System.Linq
                 }
 
                 var result = e.Current;
+
                 if (!await e.MoveNextAsync().ConfigureAwait(false))
                 {
                     return result;
@@ -102,14 +103,68 @@ namespace System.Linq
             throw new InvalidOperationException(Strings.MORE_THAN_ONE_ELEMENT);
         }
 
-        private static Task<TSource> SingleOrDefaultCore<TSource>(IAsyncEnumerable<TSource> source, Func<TSource, bool> predicate, CancellationToken cancellationToken)
+        private static async Task<TSource> SingleOrDefaultCore<TSource>(IAsyncEnumerable<TSource> source, Func<TSource, bool> predicate, CancellationToken cancellationToken)
         {
-            return source.Where(predicate).SingleOrDefault(cancellationToken);
+            var e = source.GetAsyncEnumerator(cancellationToken);
+
+            try
+            {
+                while (await e.MoveNextAsync().ConfigureAwait(false))
+                {
+                    var result = e.Current;
+
+                    if (predicate(result))
+                    {
+                        while (await e.MoveNextAsync().ConfigureAwait(false))
+                        {
+                            if (predicate(e.Current))
+                            {
+                                throw new InvalidOperationException(Strings.MORE_THAN_ONE_ELEMENT);
+                            }
+                        }
+
+                        return result;
+                    }
+                }
+
+                return default;
+            }
+            finally
+            {
+                await e.DisposeAsync().ConfigureAwait(false);
+            }
         }
 
-        private static Task<TSource> SingleOrDefaultCore<TSource>(IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate, CancellationToken cancellationToken)
+        private static async Task<TSource> SingleOrDefaultCore<TSource>(IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate, CancellationToken cancellationToken)
         {
-            return source.Where(predicate).SingleOrDefault(cancellationToken);
+            var e = source.GetAsyncEnumerator(cancellationToken);
+
+            try
+            {
+                while (await e.MoveNextAsync().ConfigureAwait(false))
+                {
+                    var result = e.Current;
+
+                    if (await predicate(result).ConfigureAwait(false))
+                    {
+                        while (await e.MoveNextAsync().ConfigureAwait(false))
+                        {
+                            if (await predicate(e.Current).ConfigureAwait(false))
+                            {
+                                throw new InvalidOperationException(Strings.MORE_THAN_ONE_ELEMENT);
+                            }
+                        }
+
+                        return result;
+                    }
+                }
+
+                return default;
+            }
+            finally
+            {
+                await e.DisposeAsync().ConfigureAwait(false);
+            }
         }
     }
 }