From bebc80ae6555e68f2f8e0fcd9271efa7460fe1d6 Mon Sep 17 00:00:00 2001
From: Jan Lohage <git@j2L4e.de>
Date: Wed, 14 Feb 2018 09:50:04 +0100
Subject: [PATCH] requery the source when there's not enough paginated data
 left on the client (after remove or update/patch)

---
 src/strategies.js | 30 ++++++++++++------------------
 src/utils.js      | 40 +++++++++++++++++++++++++++++++++++++++-
 2 files changed, 51 insertions(+), 19 deletions(-)

diff --git a/src/strategies.js b/src/strategies.js
index 8e21432..e113cda 100644
--- a/src/strategies.js
+++ b/src/strategies.js
@@ -1,14 +1,15 @@
 import { merge } from 'rxjs/observable/merge';
-import { of } from 'rxjs/observable/of';
+import { concat } from 'rxjs/observable/concat';
+
 import {
-  concat,
-  concatMap,
   concatMapTo,
   filter,
   map,
   scan
 } from 'rxjs/operators';
 
+import { retryWhenIncomplete } from './utils';
+
 module.exports = function () {
   return {
     never (source$) {
@@ -31,12 +32,10 @@ module.exports = function () {
         this.patched$
       );
 
-      return source$.pipe(
-        concat(
-          events$.pipe(
-            concatMapTo(source$)
-          )
-        ));
+      return concat(
+        source$,
+        events$.pipe(concatMapTo(source$))
+      );
     },
 
     smart (source$, options, args) {
@@ -114,15 +113,10 @@ module.exports = function () {
         )
       );
 
-      return source$.pipe(
-        concatMap(data =>
-          of(data).pipe(
-            concat(
-              events$.pipe(
-                scan((current, callback) => sortAndTrim(callback(current)), data))
-            )
-          )
-        ));
+      return concat(source$, events$).pipe(
+        scan((current, callback) => sortAndTrim(callback(current))),
+        retryWhenIncomplete(options)
+      );
     }
   };
 };
diff --git a/src/utils.js b/src/utils.js
index a94e1a9..842f4ec 100644
--- a/src/utils.js
+++ b/src/utils.js
@@ -1,6 +1,8 @@
 import sift from 'sift';
 import { _, sorter as createSorter } from '@feathersjs/commons/lib/utils';
 import { defer } from 'rxjs/observable/defer';
+import { tap } from 'rxjs/operators/tap';
+import { catchError } from 'rxjs/operators/catchError';
 
 function getSource (originalMethod, args) {
   return defer(() => originalMethod(...args));
@@ -66,11 +68,47 @@ function siftMatcher (originalQuery) {
   return sift(query);
 }
 
+const ERROR_INCOMPLETE = 'feathers-reactive: data is incomplete';
+
+function isComplete (page, options) {
+  const isPaginated = !!page[ options.dataField ];
+
+  if (isPaginated) {
+    const length = page[ options.dataField ].length;
+
+    if (page.total >= page.limit && length < page.limit) {
+      return false;
+    }
+  }
+
+  return true;
+}
+
+function retryWhenIncomplete (options) {
+  return function (source) {
+    return source.pipe(
+      tap(data => {
+        if (!isComplete(data, options)) {
+          throw new Error(ERROR_INCOMPLETE);
+        }
+      }),
+      catchError((err, caught) => {
+        if (err.message === ERROR_INCOMPLETE) {
+          return caught;
+        }
+
+        throw err;
+      })
+    );
+  };
+}
+
 Object.assign(exports, {
   sift,
   getSource,
   makeSorter,
   getOptions,
   getParamsPosition,
-  siftMatcher
+  siftMatcher,
+  retryWhenIncomplete
 });