Async variants for First, Last, and Single.

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

@@ -2,9 +2,7 @@
 // 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;
 using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -30,6 +28,16 @@ namespace System.Linq
             return First(source, predicate, CancellationToken.None);
         }
 
+        public static Task<TSource> First<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return First(source, predicate, CancellationToken.None);
+        }
+
         public static Task<TSource> First<TSource>(this IAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
         {
             if (source == null)
@@ -49,6 +57,17 @@ namespace System.Linq
                          .First(cancellationToken);
         }
 
+        public static Task<TSource> First<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate, CancellationToken cancellationToken)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return source.Where(predicate)
+                         .First(cancellationToken);
+        }
+
         public static Task<TSource> FirstOrDefault<TSource>(this IAsyncEnumerable<TSource> source)
         {
             if (source == null)
@@ -67,6 +86,16 @@ namespace System.Linq
             return FirstOrDefault(source, predicate, CancellationToken.None);
         }
 
+        public static Task<TSource> FirstOrDefault<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return FirstOrDefault(source, predicate, CancellationToken.None);
+        }
+
         public static Task<TSource> FirstOrDefault<TSource>(this IAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
         {
             if (source == null)
@@ -86,6 +115,17 @@ namespace System.Linq
                          .FirstOrDefault(cancellationToken);
         }
 
+        public static Task<TSource> FirstOrDefault<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate, CancellationToken cancellationToken)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return source.Where(predicate)
+                         .FirstOrDefault(cancellationToken);
+        }
+
         private static async Task<TSource> First_<TSource>(IAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
         {
             var list = source as IList<TSource>;

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

@@ -2,9 +2,7 @@
 // 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;
 using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -30,6 +28,16 @@ namespace System.Linq
             return Last(source, predicate, CancellationToken.None);
         }
 
+        public static Task<TSource> Last<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return Last(source, predicate, CancellationToken.None);
+        }
+
         public static Task<TSource> Last<TSource>(this IAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
         {
             if (source == null)
@@ -49,6 +57,17 @@ namespace System.Linq
                          .Last(cancellationToken);
         }
 
+        public static Task<TSource> Last<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate, CancellationToken cancellationToken)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return source.Where(predicate)
+                         .Last(cancellationToken);
+        }
+
         public static Task<TSource> LastOrDefault<TSource>(this IAsyncEnumerable<TSource> source)
         {
             if (source == null)
@@ -67,6 +86,16 @@ namespace System.Linq
             return LastOrDefault(source, predicate, CancellationToken.None);
         }
 
+        public static Task<TSource> LastOrDefault<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return LastOrDefault(source, predicate, CancellationToken.None);
+        }
+
         public static Task<TSource> LastOrDefault<TSource>(this IAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
         {
             if (source == null)
@@ -86,6 +115,17 @@ namespace System.Linq
                          .LastOrDefault(cancellationToken);
         }
 
+        public static Task<TSource> LastOrDefault<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate, CancellationToken cancellationToken)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return source.Where(predicate)
+                         .LastOrDefault(cancellationToken);
+        }
+
         private static async Task<TSource> Last_<TSource>(IAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
         {
             var last = default(TSource);

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

@@ -2,9 +2,7 @@
 // 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;
 using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -30,6 +28,15 @@ namespace System.Linq
             return Single(source, predicate, CancellationToken.None);
         }
 
+        public static Task<TSource> Single<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return Single(source, predicate, CancellationToken.None);
+        }
 
         public static Task<TSource> Single<TSource>(this IAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
         {
@@ -50,6 +57,17 @@ namespace System.Linq
                          .Single(cancellationToken);
         }
 
+        public static Task<TSource> Single<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate, CancellationToken cancellationToken)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return source.Where(predicate)
+                         .Single(cancellationToken);
+        }
+
         public static Task<TSource> SingleOrDefault<TSource>(this IAsyncEnumerable<TSource> source)
         {
             if (source == null)
@@ -68,6 +86,16 @@ namespace System.Linq
             return SingleOrDefault(source, predicate, CancellationToken.None);
         }
 
+        public static Task<TSource> SingleOrDefault<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return SingleOrDefault(source, predicate, CancellationToken.None);
+        }
+
         public static Task<TSource> SingleOrDefault<TSource>(this IAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
         {
             if (source == null)
@@ -87,6 +115,17 @@ namespace System.Linq
                          .SingleOrDefault(cancellationToken);
         }
 
+        public static Task<TSource> SingleOrDefault<TSource>(this IAsyncEnumerable<TSource> source, Func<TSource, Task<bool>> predicate, CancellationToken cancellationToken)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            return source.Where(predicate)
+                         .SingleOrDefault(cancellationToken);
+        }
+
         private static async Task<TSource> Single_<TSource>(IAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
         {
             if (source is IList<TSource> list)