+SwitchMap, +Retry(Task), +Repeat(Task), +Concat(IA<IA>), +Merge(IA<IA>)

Author: akarnokd

README.md

@@ -51,6 +51,7 @@ finally
 - `Merge` - run multiple sources at once and merge their items into a single async sequence
 - `Never` - the async sequence never produces any items and never terminates
 - `Range` - emit a range of numbers
+- `Switch` - switch between inner async sources produced by an outer async sequence
 - `Timer` - emit zero after some time delay
 - `Using` - use a resource for the duration of a generated actual `IAsyncEnumerable`
 - `Zip` - combine the next values of multiple sources via a function and emit its results
@@ -89,6 +90,7 @@ finally
 - `SkipUntil` - skip until another async sequence signals an item or completes
 - `SkipWhile` - skip items while the predicate returns true, start emitting when it turns false
 - `SwitchIfEmpty` - switch to an alternate async sequence if the main sequence turns out to be empty
+- `SwitchMap` - switch to a newer mapped-in async sequence, disposing the old one, whenever the source produces an item
 - `Take` - take at most a given number of items and stop the async sequence after that
 - `TakeLast` - take the last given number of items of the source async sequence and emit those
 - `TakeUntil` - take items from the main source until a secondary async sequence signals an item or completes

async-enumerable-dotnet-benchmark/Program.cs

@@ -23,43 +23,25 @@ static void Main(string[] args)
         {
             for (var i = 0; i < 100000; i++)
             {
-                if (i % 100 == 0)
+                if (i % 10 == 0)
                 {
                     Console.WriteLine(i);
                 }
-                var list = AsyncEnumerable.Range(1, 5)
-                    .Publish(a => 
-                        a.Take(3).MergeWith(a.Skip(3))
-                    )
-                    .ToList().GetAsyncEnumerator();
+                var list = AsyncEnumerable.Range(1, 100_000)
+                .SwitchMap(v => AsyncEnumerable.Range(v, 2))
+                .Last()
+                .GetAsyncEnumerator();
+
                 try
                 {
-                    /*
-                    var t = 0;
-                    while (!list.IsCompleted && !list.IsFaulted && !list.IsCanceled && t < 5000)
-                    {
-                        t++;
-                    }
-
-                    while (!list.IsCompleted && !list.IsFaulted && !list.IsCanceled)
-                    {
-                        await Task.Delay(1);
-                    }
-                    */
-
                     if (!list.MoveNextAsync().Result)
                     {
                         Console.WriteLine("Empty?");
                     }
 
-                    if (list.Current.Count != 5)
+                    if (list.Current != 100_001)
                     {
-                        foreach (var e in list.Current)
-                        {
-                            Console.Write(e);
-                            Console.Write(", ");
-                        }
-                        Console.WriteLine();
+                        Console.WriteLine(list.Current);
                         Console.ReadLine();
                         break;
                     }

async-enumerable-dotnet-test/AsyncEnumerableTest.cs

@@ -210,6 +210,8 @@ static AsyncEnumerableTest()
             Defaults.Add(typeof(Func<ValueTask>), (Func<ValueTask>)(() => new ValueTask()));
             Defaults.Add(typeof(Func<long, bool>), (Func<long, bool>)(v => false));
             Defaults.Add(typeof(Func<long, Exception, bool>), (Func<long, Exception, bool>)((v, w) => false));
+            Defaults.Add(typeof(Func<long, Exception, Task<bool>>), (Func<long, Exception, Task<bool>>)((v, w) => Task.FromResult(false)));
+            Defaults.Add(typeof(Func<long, Task<bool>>), (Func<long, Task<bool>>)(v => Task.FromResult(false)));
 
             Defaults.Add(typeof(Func<IAsyncEnumerable<int>, IAsyncEnumerable<int>>), (Func<IAsyncEnumerable<int>, IAsyncEnumerable<int>>)(w => w));
 

async-enumerable-dotnet-test/ConcatTest.cs

@@ -57,5 +57,22 @@ await AsyncEnumerable.Range(1, 5)
                 .ConcatWith(AsyncEnumerable.Range(4, 2))
                 .AssertResult(1, 2, 3, 4, 5);
         }
+
+        [Fact]
+        public async void Async_Normal()
+        {
+            await 
+                AsyncEnumerable.FromArray(
+                    AsyncEnumerable.Range(1, 3),
+                    AsyncEnumerable.Empty<int>(),
+                    AsyncEnumerable.FromArray(4, 5, 6, 7),
+                    AsyncEnumerable.Empty<int>(),
+                    AsyncEnumerable.Just(8),
+                    AsyncEnumerable.FromEnumerable(new[] { 9, 10 }),
+                    AsyncEnumerable.Empty<int>()
+                )
+                .Concat()
+                .AssertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+        }
     }
 }

async-enumerable-dotnet-test/MergeTest.cs

@@ -112,5 +112,23 @@ await AsyncEnumerable.Range(1, 5)
             }
         }
 
+
+        [Fact]
+        public async void Async_Normal()
+        {
+            await
+                AsyncEnumerable.FromArray(
+                    AsyncEnumerable.Range(1, 3),
+                    AsyncEnumerable.Empty<int>(),
+                    AsyncEnumerable.FromArray(4, 5, 6, 7),
+                    AsyncEnumerable.Empty<int>(),
+                    AsyncEnumerable.Just(8),
+                    AsyncEnumerable.FromEnumerable(new[] { 9, 10 }),
+                    AsyncEnumerable.Empty<int>()
+                )
+                .Merge()
+                .AssertResultSet(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+        }
+
     }
 }

async-enumerable-dotnet-test/RepeatTest.cs

@@ -4,6 +4,7 @@
 
 using Xunit;
 using async_enumerable_dotnet;
+using System.Threading.Tasks;
 
 namespace async_enumerable_dotnet_test
 {
@@ -34,5 +35,15 @@ await AsyncEnumerable.Range(1, 2)
                 .AssertResult(1, 2, 1, 2, 1, 2);
         }
 
+        [Fact]
+        public async void Limited_Condition_Task()
+        {
+            await AsyncEnumerable.Range(1, 2)
+                .Repeat(async n => {
+                    await Task.Delay(100);
+                    return n < 2;
+                })
+                .AssertResult(1, 2, 1, 2, 1, 2);
+        }
     }
 }

async-enumerable-dotnet-test/RetryTest.cs

@@ -5,6 +5,7 @@
 using System;
 using Xunit;
 using async_enumerable_dotnet;
+using System.Threading.Tasks;
 
 namespace async_enumerable_dotnet_test
 {
@@ -37,5 +38,17 @@ await AsyncEnumerable.Range(1, 2)
                 .Retry((idx, ex) => idx < 2)
                 .AssertFailure(typeof(InvalidOperationException), 1, 2, 1, 2, 1, 2);
         }
+
+        [Fact]
+        public async void Retry_Condition_Task()
+        {
+            await AsyncEnumerable.Range(1, 2)
+                .ConcatWith(AsyncEnumerable.Error<int>(new InvalidOperationException()))
+                .Retry(async (idx, ex) => {
+                    await Task.Delay(100);
+                    return idx < 2;
+                })
+                .AssertFailure(typeof(InvalidOperationException), 1, 2, 1, 2, 1, 2);
+        }
     }
 }

async-enumerable-dotnet-test/SwitchMapTest.cs

@@ -0,0 +1,83 @@
+// Copyright (c) David Karnok & Contributors.
+// Licensed under the Apache 2.0 License.
+// See LICENSE file in the project root for full license information.
+
+using Xunit;
+using async_enumerable_dotnet;
+using System;
+
+namespace async_enumerable_dotnet_test
+{
+    public class SwitchMapTest
+    {
+        [Fact]
+        public async void Empty()
+        {
+            await AsyncEnumerable.Empty<int>()
+                .SwitchMap(v => AsyncEnumerable.Range(1, 5))
+                .AssertResult();
+        }
+
+        [Fact]
+        public async void Single()
+        {
+            await AsyncEnumerable.Just(2)
+                .SwitchMap(v => AsyncEnumerable.Range(v, 5))
+                .AssertResult(2, 3, 4, 5, 6);
+        }
+
+        [Fact]
+        public async void Many_Switched()
+        {
+            await AsyncEnumerable.Range(1, 5)
+                .SwitchMap(v => AsyncEnumerable.Range(v * 10, 5))
+                .Last()
+                .AssertResult(54);
+        }
+
+
+        [Fact]
+        public async void Many_Switched_Lots()
+        {
+            await AsyncEnumerable.Range(1, 1000)
+                .SwitchMap(v => AsyncEnumerable.Range(v * 1000, 100))
+                .Last()
+                .AssertResult(1_000_099);
+        }
+
+        [Fact]
+        public async void Many_Switched_Lots_2()
+        {
+            await AsyncEnumerable.Range(1, 100_000)
+                .SwitchMap(v => AsyncEnumerable.Range(v, 2))
+                .Last()
+                .AssertResult(100_001);
+        }
+
+        [Fact]
+        public async void Error_Outer()
+        {
+            await AsyncEnumerable.Just(2).ConcatWith(AsyncEnumerable.Error<int>(new InvalidOperationException()))
+                .SwitchMap(v => AsyncEnumerable.Range(v, 5))
+                .AssertFailure(typeof(InvalidOperationException), 2, 3, 4, 5, 6);
+        }
+
+        [Fact]
+        public async void Error_Inner()
+        {
+            await AsyncEnumerable.Just(2)
+                .SwitchMap(v => AsyncEnumerable.Range(v, 5)
+                    .ConcatWith(AsyncEnumerable.Error<int>(new InvalidOperationException()))
+                )
+                .AssertFailure(typeof(InvalidOperationException), 2, 3, 4, 5, 6);
+        }
+
+        [Fact]
+        public async void Mapper_Crash()
+        {
+            await AsyncEnumerable.Just(2)
+                .SwitchMap<int, int>(v => throw new InvalidOperationException())
+                .AssertFailure(typeof(InvalidOperationException));
+        }
+    }
+}

async-enumerable-dotnet/AsyncEnumerable.cs

@@ -921,7 +921,7 @@ public static IAsyncEnumerable<T> Repeat<T>(this IAsyncEnumerable<T> source, lon
         /// <typeparam name="T">The element type.</typeparam>
         /// <param name="source">The source async sequence to repeatedly relay items of.</param>
         /// <param name="condition">The function called when the current run completes with
-        /// the current run index (zero-based) and should return to repeat the sequence once more, false to end it.</param>
+        /// the current run index (zero-based) and should return true to repeat the sequence once more, false to end it.</param>
         /// <returns>The new IAsyncEnumerable instance.</returns>
         public static IAsyncEnumerable<T> Repeat<T>(this IAsyncEnumerable<T> source, Func<long, bool> condition)
         {
@@ -930,6 +930,21 @@ public static IAsyncEnumerable<T> Repeat<T>(this IAsyncEnumerable<T> source, Fun
             return new Repeat<T>(source, long.MaxValue, condition);
         }
 
+        /// <summary>
+        /// Repeatedly relay the source async sequence, once it completes the previous time, when the function returns true, ending the sequence otherwise.
+        /// </summary>
+        /// <typeparam name="T">The element type.</typeparam>
+        /// <param name="source">The source async sequence to repeatedly relay items of.</param>
+        /// <param name="condition">The function called when the current run completes with
+        /// the current run index (zero-based) and should return a task with true to repeat the sequence once more, false to end it.</param>
+        /// <returns>The new IAsyncEnumerable instance.</returns>
+        public static IAsyncEnumerable<T> Repeat<T>(this IAsyncEnumerable<T> source, Func<long, Task<bool>> condition)
+        {
+            RequireNonNull(source, nameof(source));
+            RequireNonNull(condition, nameof(condition));
+            return new RepeatTask<T>(source, condition);
+        }
+
         /// <summary>
         /// Retry a possibly failing async sequence, optionally a limited number of times before giving up.
         /// </summary>
@@ -957,6 +972,20 @@ public static IAsyncEnumerable<T> Retry<T>(this IAsyncEnumerable<T> source, Func
             return new Retry<T>(source, long.MaxValue, condition);
         }
 
+        /// <summary>
+        /// Retry a possibly failing async sequence if the condition returns true for the Exception and/or retry index.
+        /// </summary>
+        /// <typeparam name="T">The element type of the source async sequence.</typeparam>
+        /// <param name="source">The source async sequence that could fail and should be repeated.</param>
+        /// <param name="condition">Called when the sequence fails with the retry index (zero-based) and last failure Exception and should return a task with true if the sequence should be retried.</param>
+        /// <returns>The new IAsyncEnumerable instance.</returns>
+        public static IAsyncEnumerable<T> Retry<T>(this IAsyncEnumerable<T> source, Func<long, Exception, Task<bool>> condition)
+        {
+            RequireNonNull(source, nameof(source));
+            RequireNonNull(condition, nameof(condition));
+            return new RetryTask<T>(source, condition);
+        }
+
         /// <summary>
         /// Skips the last given number of items from the source
         /// async sequence.
@@ -1442,5 +1471,64 @@ public static IAsyncEnumerable<TResult> Replay<TSource, TResult>(this IAsyncEnum
             RequireNonNull(func, nameof(func));
             return new Replay<TSource, TResult>(source, func);
         }
+
+        /// <summary>
+        /// Switches to a newer source async sequence, disposing the old one,
+        /// when the source async sequence produces
+        /// a value and is mapped to an async sequence via a function.
+        /// </summary>
+        /// <typeparam name="TSource">The source value type.</typeparam>
+        /// <typeparam name="TResult">The result value type.</typeparam>
+        /// <param name="source">The source to map into async sequences and switch between them.</param>
+        /// <param name="mapper">The function that receives the source item and should
+        /// return an async sequence to be run.</param>
+        /// <returns>The new IAsyncEnumerable sequence.</returns>
+        public static IAsyncEnumerable<TResult> SwitchMap<TSource, TResult>(this IAsyncEnumerable<TSource> source, Func<TSource, IAsyncEnumerable<TResult>> mapper)
+        {
+            RequireNonNull(source, nameof(source));
+            RequireNonNull(mapper, nameof(mapper));
+            return new SwitchMap<TSource, TResult>(source, mapper);
+        }
+
+        /// <summary>
+        /// Switches to a newer source async sequence, disposing the old one,
+        /// when the source async sequence produces a new inner async sequence.
+        /// </summary>
+        /// <typeparam name="TSource">The source value type.</typeparam>
+        /// <param name="sources">The async sequence of async sequences to switch between.</param>
+        /// <returns>The new IAsyncEnumerable sequence.</returns>
+        public static IAsyncEnumerable<TSource> Switch<TSource>(this IAsyncEnumerable<IAsyncEnumerable<TSource>> sources)
+        {
+            RequireNonNull(sources, nameof(sources));
+            return sources.SwitchMap(v => v);
+        }
+
+        /// <summary>
+        /// Runs the inner async sequences, produced by an outer async sequence,
+        /// one after the other and relays their items.
+        /// </summary>
+        /// <typeparam name="TSource">The value type.</typeparam>
+        /// <param name="sources">The async sequence of async sequences to concatenate.</param>
+        /// <returns>The new IAsyncEnumerable sequence.</returns>
+        public static IAsyncEnumerable<TSource> Concat<TSource>(this IAsyncEnumerable<IAsyncEnumerable<TSource>> sources)
+        {
+            RequireNonNull(sources, nameof(sources));
+            return sources.ConcatMap(v => v);
+        }
+
+        /// <summary>
+        /// Runs the inner async sequences, produced by an outer async sequence,
+        /// some or all concurrently and relays their items single a serialized async sequence.
+        /// </summary>
+        /// <typeparam name="TSource">The value type.</typeparam>
+        /// <param name="sources">The async sequence of async sequences to merge at once.</param>
+        /// <param name="maxConcurrency">The maximum number of inner sequences to run at once.</param>
+        /// <param name="prefetch">The number of items to prefetch from each inner async sequence.</param>
+        /// <returns>The new IAsyncEnumerable instance.</returns>
+        public static IAsyncEnumerable<TSource> Merge<TSource>(this IAsyncEnumerable<IAsyncEnumerable<TSource>> sources, int maxConcurrency = int.MaxValue, int prefetch = 32)
+        {
+            RequireNonNull(sources, nameof(sources));
+            return sources.FlatMap(v => v, maxConcurrency, prefetch);
+        }
     }
 }