2 files changed, 64 insertions, 0 deletions
diff --git a/dsa/FireBase/Extensions/ObservableExtensions.cs b/dsa/FireBase/Extensions/ObservableExtensions.cs
new file mode 100644
index 0000000..0a672d7
--- /dev/null
+++ b/dsa/FireBase/Extensions/ObservableExtensions.cs
@@ -0,0 +1,41 @@
+using System;
+using System.Reactive.Linq;
+
+namespace Firebase.Database.Extensions
+{
+    public static class ObservableExtensions
+    {
+        /// <summary>
+        ///     Returns a cold observable which retries (re-subscribes to) the source observable on error until it successfully
+        ///     terminates.
+        /// </summary>
+        /// <param name="source">The source observable.</param>
+        /// <param name="dueTime">How long to wait between attempts.</param>
+        /// <param name="retryOnError">A predicate determining for which exceptions to retry. Defaults to all</param>
+        /// <returns>
+        ///     A cold observable which retries (re-subscribes to) the source observable on error up to the
+        ///     specified number of times or until it successfully terminates.
+        /// </returns>
+        public static IObservable<T> RetryAfterDelay<T, TException>(
+            this IObservable<T> source,
+            TimeSpan dueTime,
+            Func<TException, bool> retryOnError)
+            where TException : Exception
+        {
+            var attempt = 0;
+
+            return Observable.Defer(() =>
+                {
+                    return (++attempt == 1 ? source : source.DelaySubscription(dueTime))
+                        .Select(item => new Tuple<bool, T, Exception>(true, item, null))
+                        .Catch<Tuple<bool, T, Exception>, TException>(e => retryOnError(e)
+                            ? Observable.Throw<Tuple<bool, T, Exception>>(e)
+                            : Observable.Return(new Tuple<bool, T, Exception>(false, default(T), e)));
+                })
+                .Retry()
+                .SelectMany(t => t.Item1
+                    ? Observable.Return(t.Item2)
+                    : Observable.Throw<T>(t.Item3));
+        }
+    }
+}
+\ No newline at end of file
diff --git a/dsa/FireBase/Extensions/TaskExtensions.cs b/dsa/FireBase/Extensions/TaskExtensions.cs
new file mode 100644
index 0000000..c955b3a
--- /dev/null
+++ b/dsa/FireBase/Extensions/TaskExtensions.cs
@@ -0,0 +1,23 @@
+using System;
+using System.Threading.Tasks;
+
+namespace Firebase.Database.Extensions
+{
+    public static class TaskExtensions
+    {
+        /// <summary>
+        ///     Instead of unwrapping <see cref="AggregateException" /> it throws it as it is.
+        /// </summary>
+        public static async Task WithAggregateException(this Task source)
+        {
+            try
+            {
+                await source.ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                throw source.Exception ?? ex;
+            }
+        }
+    }
+}
+\ No newline at end of file