Merge pull request #753 from akarnokd/IxRetryImprovements

Ix: Implement the Retry() operator directly

Ix.NET/Source/Benchmarks.System.Interactive/Program.cs

@@ -20,6 +20,7 @@ namespace Benchmarks.System.Interactive
                 typeof(BufferCountBenchmark),
                 typeof(IgnoreElementsBenchmark),
                 typeof(DeferBenchmark),
+                typeof(RetryBenchmark),
                 typeof(MinMaxBenchmark),
             });
 

Ix.NET/Source/Benchmarks.System.Interactive/RetryBenchmark.cs

@@ -0,0 +1,34 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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 BenchmarkDotNet.Attributes;
+
+namespace Benchmarks.System.Interactive
+{
+    [MemoryDiagnoser]
+    public class RetryBenchmark
+    {
+        [Params(1, 10, 100, 1000, 10000, 100000, 1000000)]
+        public int N;
+        private int _store;
+
+        [Benchmark]
+        public void Finite()
+        {
+            Enumerable.Range(1, N).Retry(100)
+                .Subscribe(v => Volatile.Write(ref _store, v));
+        }
+
+        [Benchmark]
+        public void Infinite()
+        {
+            Enumerable.Range(1, N).Retry()
+                .Subscribe(v => Volatile.Write(ref _store, v));
+        }
+    }
+}

Ix.NET/Source/System.Interactive/Retry.cs

@@ -21,8 +21,7 @@ namespace System.Linq
                 throw new ArgumentNullException(nameof(source));
             }
 
-            return new[] { source }.Repeat()
-                                 .Catch();
+            return RetryInfinite(source);
         }
 
         /// <summary>
@@ -45,8 +44,94 @@ namespace System.Linq
                 throw new ArgumentOutOfRangeException(nameof(retryCount));
             }
 
-            return new[] { source }.Repeat(retryCount)
-                                 .Catch();
+            return RetryFinite(source, retryCount);
+        }
+
+        private static IEnumerable<TSource> RetryInfinite<TSource>(IEnumerable<TSource> source)
+        {
+            while (true)
+            {
+                var enumerator = default(IEnumerator<TSource>);
+                try
+                {
+                    enumerator = source.GetEnumerator();
+                }
+                catch
+                {
+                    continue;
+                }
+
+                using (enumerator)
+                {
+                    while (true)
+                    {
+                        var v = default(TSource);
+
+                        try
+                        {
+                            if (!enumerator.MoveNext())
+                            {
+                                yield break;
+                            }
+                            v = enumerator.Current;
+                        }
+                        catch
+                        {
+                            break;
+                        }
+
+                        yield return v;
+                    }
+                }
+            }
+        }
+
+        private static IEnumerable<TSource> RetryFinite<TSource>(IEnumerable<TSource> source, int retryCount)
+        {
+            var lastException = default(Exception);
+
+            for (var i = 0; i < retryCount; i++)
+            {
+                var enumerator = default(IEnumerator<TSource>);
+                try
+                {
+                    enumerator = source.GetEnumerator();
+                }
+                catch (Exception ex)
+                {
+                    lastException = ex;
+                    continue;
+                }
+
+                using (enumerator)
+                {
+                    while (true)
+                    {
+                        var v = default(TSource);
+
+                        try
+                        {
+                            if (!enumerator.MoveNext())
+                            {
+                                yield break;
+                            }
+                            v = enumerator.Current;
+                        }
+                        catch (Exception ex)
+                        {
+                            lastException = ex;
+                            break;
+                        }
+
+                        yield return v;
+                    }
+                }
+            }
+
+            if (lastException != null)
+            {
+                throw lastException;
+            }
         }
     }
-}
+}