Merge pull request step-up-labs#212 from cabauman/colt/pull-async-fix

PullAsync fix

Author: bezysoftware

src/Firebase/Extensions/ObservableExtensions.cs

@@ -11,28 +11,39 @@ public static class ObservableExtensions
         /// <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>
+        /// <param name="retryCount">The number of attempts of running the source observable before failing.</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)
+            Func<TException, bool> retryOnError,
+            int? retryCount = null)
             where TException: Exception
         {
             int attempt = 0;
 
-            return Observable.Defer(() =>
+            var pipeline = 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
+            });
+
+            if (retryCount.HasValue)
+            {
+                pipeline = pipeline.Retry(retryCount.Value);
+            }
+            else
+            {
+                pipeline = pipeline.Retry();
+            }
+
+            return pipeline.SelectMany(t => t.Item1
                 ? Observable.Return(t.Item2)
                 : Observable.Throw<T>(t.Item3));
         }

src/Firebase/Offline/RealtimeDatabase.cs

@@ -149,14 +149,17 @@ public void Pull(string key, int priority = 1)
         /// <summary>
         /// Fetches everything from the remote database.
         /// </summary>
-        public async Task PullAsync()
+        /// <param name="retryCount">
+        /// The number of attempts of running the source observable before failing.
+        /// null means infinite retries, 0 means it won't execute at all, and 1 means it will only try once.
+        /// </param>
+        public async Task PullAsync(int? retryCount = null)
         {
-            var existingEntries = await this.childQuery
-                .OnceAsync<T>()
-                .ToObservable()
+            var existingEntries = await Observable.Defer(() => this.childQuery.OnceAsync<T>().ToObservable())
                 .RetryAfterDelay<IReadOnlyCollection<FirebaseObject<T>>, FirebaseException>(
                     this.childQuery.Client.Options.SyncPeriod,
-                    ex => ex.StatusCode == System.Net.HttpStatusCode.OK) // OK implies the request couldn't complete due to network error. 
+                    ex => ex.StatusCode == System.Net.HttpStatusCode.OK, // OK implies the request couldn't complete due to network error. 
+                    retryCount)
                 .Select(e => this.ResetDatabaseFromInitial(e, false))
                 .SelectMany(e => e)
                 .Do(e =>