Pagination

.gitignore

@@ -30,3 +30,6 @@ node_modules
 # The compiled/babelified modules
 lib/
 tmp/
+
+# VSCode config
+.vscode

package.json

@@ -31,7 +31,7 @@
     "release:patch": "npm version patch && npm publish",
     "release:minor": "npm version minor && npm publish",
     "release:major": "npm version major && npm publish",
-    "compile": "rm -rf lib/ && babel -d lib/ src/",
+    "compile": "rimraf lib/ && babel -d lib/ src/",
     "watch": "babel --watch -d lib/ src/",
     "jshint": "jshint src/. test/. --config",
     "mocha": "mocha test/ --compilers js:babel-core/register",
@@ -43,16 +43,20 @@
   "dependencies": {
     "debug": "^2.2.0",
     "feathers-commons": "^0.7.1",
-    "rx": "^4.0.7"
+    "rxjs": "^5.0.0-beta.6"
   },
   "devDependencies": {
     "babel-cli": "^6.4.5",
     "babel-core": "^6.4.5",
     "babel-plugin-add-module-exports": "^0.1.2",
+    "babel-plugin-transform-es2015-modules-commonjs": "^6.7.7",
+    "babel-plugin-transform-function-bind": "^6.5.2",
     "babel-preset-es2015": "^6.3.13",
     "feathers": "^2.0.0",
+    "feathers-hooks": "^1.5.2",
     "feathers-memory": "^0.7.0",
     "jshint": "^2.9.1",
-    "mocha": "^2.3.4"
+    "mocha": "^2.3.4",
+    "rimraf": "^2.5.2"
   }
 }

src/index.js

@@ -1,12 +1,17 @@
-import Rx from 'rx';
+import Rx from 'rxjs/Rx';
+
 import reactiveResource from './resource';
 import reactiveList from './list';
 
 const debug = require('debug')('feathers-rx');
 
-export default function(options) {
+function FeathersRx(options) {
+
   options = Object.assign({
-    id: 'id',
+    idField: 'id',
+    dataField: 'data',
+    // Whether to requery service when a change is detected
+    strategy: reactiveList.strategy.never,
     // The merging strategy
     merge(current, eventData) {
       return Object.assign({}, current, eventData);
@@ -22,10 +27,10 @@ export default function(options) {
       patched: Rx.Observable.fromEvent(service, 'patched'),
       removed: Rx.Observable.fromEvent(service, 'removed')
     };
-    const resourceMethod = reactiveResource(events, options);
 
     app.methods.forEach(method => {
       if(method !== 'find' && typeof service[method] === 'function') {
+        const resourceMethod = reactiveResource(events, method, options);
         mixin[method] = resourceMethod;
       }
     });
@@ -37,11 +42,25 @@ export default function(options) {
     service.mixin(mixin);
   };
 
+  const serviceMixin = function (service) {
+    const mixin = {};
+    mixin.rx = (options) => {
+      service._rx = options? options: {};
+      return service;
+    };
+    service.mixin(mixin);
+  };
+
   return function() {
     debug('Initializing feathers-rx plugin');
 
     const app = this;
 
     app.mixins.push(mixin);
+    app.mixins.push(serviceMixin);
   };
 }
+
+FeathersRx.strategy = reactiveList.strategy;
+
+export default FeathersRx;

src/list.js

@@ -1,11 +1,18 @@
-import Rx from 'rx';
+import Rx from 'rxjs/Rx';
+import 'rxjs/add/operator/exhaustMap';
+
 import { promisify } from './utils';
 import { sorter as createSorter, matcher } from 'feathers-commons/lib/utils';
 
-export default function(events, options) {
-  return function(params) {
+function List (events, options) {
+
+  return function (params) {
+
     const query = Object.assign({}, params.query);
     const result = this._super.apply(this, arguments);
+    const inputArgs = arguments;
+    params = params ? params : {}; // No params
+    options = Object.assign(options, this._rx, params.rx);
 
     if(typeof result.then !== 'function') {
       return result;
@@ -19,30 +26,81 @@ export default function(events, options) {
     // The sort function (if $sort is set)
     const sorter = query.$sort ? createSorter(query.$sort) : null;
 
-    const stream = source.concat(source.flatMapFirst(data =>
-      Rx.Observable.merge(
-        events.created.filter(matches).map(eventData =>
-          items => items.concat(eventData)
-        ),
-        events.removed.map(eventData =>
-          items => items.filter(current => eventData[options.id] !== current[options.id])
-        ),
-        updaters.map(eventData =>
-          items => items.map(current => {
-            if(eventData[options.id] === current[options.id]) {
-              return options.merge(current, eventData);
-            }
-
-            return current;
-          }).filter(matches)
-        )
-      ).scan((current, callback) => {
-        const result = callback(current);
-
-        return sorter ? result.sort(sorter) : result;
-      }, data)
-    ));
+    let stream;
+
+    const sortAndTrim = (result) => {
+      const isPaginated = !!result[options.dataField];
+      if (isPaginated) {
+        result.data = (sorter ? result.data.sort(sorter) : result.data).slice(0,result.limit);
+      } else {
+        result = (sorter ? result.sort(sorter) : result).slice(0,result.limit);
+      }
+      return result;
+    };
+
+    if (options.strategy === List.strategy.never) {
+
+      stream = source.concat(source.exhaustMap(data =>
+        Rx.Observable.merge(
+          events.created.filter(matches).map(eventData =>
+            items => items.concat(eventData)
+          ),
+          events.removed.map(eventData =>
+            items => items.filter(current => eventData[options.idField] !== current[options.idField])
+          ),
+          updaters.map(eventData =>
+            items => items.map(current => {
+              if(eventData[options.idField] === current[options.idField]) {
+                return options.merge(current, eventData);
+              }
+
+              return current;
+            }).filter(matches)
+          )
+        ).scan((current, callback) => {
+          const isPaginated = !!current[options.dataField];
+          if (isPaginated) {
+            current.data = callback(current.data);
+          } else {
+            current = callback(current);
+          }
+          return sortAndTrim(current);
+        }, data)
+      ));
+
+    } else if (options.strategy === List.strategy.always) {
+
+      stream = source.concat(source.exhaustMap(() =>
+        Rx.Observable.merge(
+          events.created.filter(matches),
+          events.removed,
+          updaters
+        ).flatMap(() => {
+          const result = this.find.apply(this, inputArgs);
+          const source = Rx.Observable.fromPromise(result);
+          if(typeof result.then !== 'function') {
+            return Rx.Observable.of(result);
+          }
+
+          return source.map((result) => {
+            return sortAndTrim(result);
+          });
+        })
+      ));
+
+    } else {
+      throw 'Unsupported feathers-rx strategy type.';
+    }
 
     return promisify(stream, result);
   };
 }
+
+List.strategy = {
+  always: {},
+  never: {},
+  // TODO: Jack
+  // smart: {}
+};
+
+export default List;

src/resource.js

@@ -1,19 +1,32 @@
-import Rx from 'rx';
+import Rx from 'rxjs/Rx';
+import 'rxjs/add/operator/exhaustMap';
+
 import { promisify } from './utils';
 
-export default function(events, options) {
+// The position of the params parameters for a service method so that we can extend them
+// default is 1
+export const paramsPositions = {
+  update: 2,
+  patch: 2
+};
+
+export default function(events, method, options) {
   return function() {
     const result = this._super.apply(this, arguments);
+    let position = typeof paramsPositions[method] !== 'undefined' ?
+      paramsPositions[method] : 1;
+    let params = arguments[position] || {};
+    options = Object.assign(options, this._rx, params.rx);
 
     // We only support promises
     if(typeof result.then !== 'function') {
       return result;
     }
 
     const source = Rx.Observable.fromPromise(result);
-    const stream = source.concat(source.flatMapFirst(data => {
+    const stream = source.concat(source.exhaustMap(data => {
       // Filter only data with the same id
-      const filter = current => current[options.id] === data[options.id];
+      const filter = current => current[options.idField] === data[options.idField];
       // `removed` events get special treatment
       const filteredRemoves = events.removed.filter(filter);
       // `created`, `updated` and `patched`