Allow changefeed event data to be processed by after hooks (#75)

* Adding a breaking test for events not sending after hook data

* Allow changefeed data to be processed through after hooks

package.json

@@ -55,6 +55,7 @@
     "debug": "^2.2.0",
     "feathers-commons": "^0.8.4",
     "feathers-errors": "^2.0.2",
+    "feathers-hooks": "^1.8.1",
     "feathers-query-filters": "^2.0.0",
     "uberproto": "^1.2.0"
   },
@@ -67,6 +68,7 @@
     "chai": "^3.5.0",
     "eslint-if-supported": "^1.0.1",
     "feathers": "^2.0.0",
+    "feathers-hooks": "^1.8.1",
     "feathers-rest": "^1.2.4",
     "feathers-service-tests": "^0.9.0",
     "feathers-socketio": "^1.3.3",

src/index.js

@@ -1,7 +1,8 @@
 import Proto from 'uberproto';
 import filter from 'feathers-query-filters';
 import errors from 'feathers-errors';
-import { _, select } from 'feathers-commons';
+import { _, select, hooks } from 'feathers-commons';
+import { processHooks, getHooks } from 'feathers-hooks/lib/commons';
 import { createFilter } from './parse';
 
 const BASE_EVENTS = ['created', 'updated', 'patched', 'removed'];
@@ -24,7 +25,7 @@ class Service {
     }
 
     // if no options.db on service use default from pool master
-    if(!options.db) {
+    if (!options.db) {
       options.db = options.r._poolMaster._options.db;
     }
 
@@ -256,26 +257,78 @@ class Service {
       }).then(select(params, this.id));
   }
 
-  setup () {
-    if (this.watch) {
-      this._cursor = this.table.changes().run().then(cursor => {
-        cursor.each((error, data) => {
-          if (error || typeof this.emit !== 'function') {
-            return;
-          }
-          if (data.old_val === null) {
-            this.emit('created', data.new_val);
-          } else if (data.new_val === null) {
-            this.emit('removed', data.old_val);
-          } else {
-            this.emit('updated', data.new_val);
-            this.emit('patched', data.new_val);
+  watchChangefeeds (app) {
+    if (!this.watch || this._cursor) {
+      return this._cursor;
+    }
+
+    let runHooks = (method, data) => Promise.resolve({
+      result: data
+    });
+
+    if (this.__hooks) { // If the hooks plugin is enabled
+      // This is necessary because the data coming directly from the
+      // change feeds does not run through `after` hooks by default
+      // so we have to do it manually
+      runHooks = (method, data) => {
+        const service = this;
+        const args = [ { query: {}, provider: 'rethinkdb' } ];
+        const hookData = {
+          app,
+          service,
+          result: data,
+          get path () {
+            return Object.keys(app.services)
+              .find(path => app.services[path] === service);
           }
-        });
+        };
 
-        return cursor;
-      });
+        // Add `data` to arguments
+        if (method === 'create' || method === 'update' || method === 'patch') {
+          args.unshift(data);
+        }
+
+        // `id` for update, patch and remove
+        if (method === 'update' || method === 'patch' || method === 'remove') {
+          args.unshift(data[this.id]);
+        }
+
+        const hookObject = hooks.hookObject(method, 'after', args, hookData);
+        const hookChain = getHooks(app, this, 'after', method);
+
+        return processHooks.call(this, hookChain, hookObject);
+      };
     }
+
+    this._cursor = this.table.changes().run().then(cursor => {
+      cursor.each((error, data) => {
+        if (error || typeof this.emit !== 'function') {
+          return;
+        }
+        // For each case, run through processHooks first,
+        // then emit the event
+        if (data.old_val === null) {
+          runHooks('create', data.new_val)
+            .then(hook => this.emit('created', hook.result));
+        } else if (data.new_val === null) {
+          runHooks('remove', data.old_val)
+            .then(hook => this.emit('removed', hook.result));
+        } else {
+          runHooks('patch', data.new_val).then(hook => {
+            this.emit('updated', hook.result);
+            this.emit('patched', hook.result);
+          });
+        }
+      });
+
+      return cursor;
+    });
+
+    return this._cursor;
+  }
+
+  setup (app) {
+    this.watchChangefeeds(app);
   }
 }
 

test/index.test.js

@@ -2,6 +2,8 @@ import chai from 'chai';
 import { base, example } from 'feathers-service-tests';
 import feathers from 'feathers';
 import errors from 'feathers-errors';
+import hooks from 'feathers-hooks';
+
 import rethink from 'rethinkdbdash';
 import service from '../src';
 
@@ -43,6 +45,7 @@ const numberService = {
 };
 
 const app = feathers()
+  .configure(hooks())
   .use('/people', service({
     Model: r,
     name: 'people',
@@ -58,6 +61,12 @@ const app = feathers()
   }).extend(numberService));
 const people = app.service('people');
 
+people.hooks({
+  after (hook) {
+    hook.result.test = 'testing';
+  }
+});
+
 describe('feathers-rethinkdb', () => {
   before(() => {
     return r.dbList().contains('feathers') // create db if not exists
@@ -89,6 +98,33 @@ describe('feathers-rethinkdb', () => {
     expect(typeof 1).to.equal('number');
   });
 
+  it('after hooks run and get send with events', done => {
+    const name = 'Hooks tester';
+    const service = app.service('people');
+
+    service.once('created', person => {
+      try {
+        expect(person.test).to.equal('testing', 'Hook property got set');
+      } catch (e) {
+        done(e);
+      }
+    });
+
+    service.once('removed', person => {
+      try {
+        expect(person.test).to.equal('testing', 'Hook property got set');
+        done();
+      } catch (e) {
+        done(e);
+      }
+    });
+
+    service.create({
+      name,
+      age: 1
+    }).then(person => service.remove(person.id));
+  });
+
   describe('common tests', () => {
     base(app, errors);
     base(app, errors, 'people-customid', 'customid');