Implement subscription filter option

.travis.yml

@@ -1,8 +1,7 @@
 language: node_js
 sudo: false
 node_js:
-  - "8"
-  - "10"
+  - "12"
 services:
   - redis
   - mongodb

examples/a.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const Hapi = require('hapi')
+const Hapi = require('@hapi/hapi')
 const routes = require('./routes')
 
 async function init () {

examples/b.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const Hapi = require('hapi')
+const Hapi = require('@hapi/hapi')
 const routes = require('./routes')
 
 async function init () {

examples/pub.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const Nes = require('nes')
+const Nes = require('@hapi/nes')
 
 async function pub () {
   const client = new Nes.Client('ws://localhost:3001')

examples/routes.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const Nes = require('nes')
+const Nes = require('@hapi/nes')
 const Multines = require('..')
 
 module.exports = async (server) => {

examples/sub.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const Nes = require('nes')
+const Nes = require('@hapi/nes')
 
 async function sub () {
   const client = new Nes.Client('ws://localhost:3000')

multines.js

@@ -7,8 +7,16 @@ const mongodb = require('mqemitter-mongodb')
 
 const kDeliver = Symbol.for('deliver')
 
-function buildDeliver (socket, topic) {
+function buildDeliver (socket, topic, filterOpts) {
   return async function deliver (message, done) {
+    const filterArgs = {
+      socket,
+      credentials: socket.auth.credentials,
+      params: filterOpts.params
+    }
+
+    if (await filterOpts.filter('/' + topic, topic === message.topic ? message.body : message, filterArgs) === false) return
+
     if (topic === message.topic) {
       await socket.publish('/' + topic, message.body)
     } else {
@@ -42,14 +50,15 @@ function getMq (options) {
 }
 
 async function register (server, options) {
-  server.dependency('nes')
+  server.dependency('@hapi/nes')
   const mq = getMq(options)
 
   server.decorate('server', 'subscriptionFar', (path, options) => {
     options = options || {}
 
     const wrapSubscribe = options.onSubscribe || (async (socket, path, params) => null)
     const wrapUnsubscribe = options.onUnsubscribe || (async (socket, path, params) => null)
+    const filter = options.filter || (async (path, message, options) => true)
 
     options.onSubscribe = async (socket, path, params) => {
       const deliverMap = socket[kDeliver] || {}
@@ -58,7 +67,7 @@ async function register (server, options) {
       const topic = path.replace(/^\//, '')
 
       if (!deliverMap[path]) {
-        deliverMap[path] = buildDeliver(socket, topic)
+        deliverMap[path] = buildDeliver(socket, topic, { filter, params })
       }
 
       await wrapSubscribe(socket, path, params)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "multines",
-  "version": "1.0.0",
+  "version": "2.0.0",
   "description": "Multi-process nes backend, turn nes into a fully scalable solution",
   "main": "multines.js",
   "scripts": {
@@ -27,16 +27,16 @@
   },
   "homepage": "https://github.com/mcollina/multines#readme",
   "devDependencies": {
-    "code": "^5.2.0",
-    "hapi": "^17.5.0",
-    "lab": "^15.4.5",
-    "nes": "^8.1.0",
+    "@hapi/code": "^8.0.1",
+    "@hapi/hapi": "^19.0.0",
+    "@hapi/lab": "^22.0.3",
+    "@hapi/nes": "^12.0.0",
     "pre-commit": "^1.1.2",
-    "standard": "^11.0.0"
+    "standard": "^14.3.1"
   },
   "dependencies": {
-    "mqemitter": "^2.2.0",
-    "mqemitter-mongodb": "^3.0.2",
-    "mqemitter-redis": "^2.3.0"
+    "mqemitter": "^4.1.3",
+    "mqemitter-mongodb": "^7.0.0",
+    "mqemitter-redis": "^4.0.2"
   }
 }

test.js

@@ -1,10 +1,10 @@
 'use strict'
 
 const mqemitter = require('mqemitter')
-const Code = require('code')
-const Lab = require('lab')
-const Hapi = require('hapi')
-const Nes = require('nes')
+const Code = require('@hapi/code')
+const Lab = require('@hapi/lab')
+const Hapi = require('@hapi/hapi')
+const Nes = require('@hapi/nes')
 const Multines = require('.')
 
 const lab = exports.lab = Lab.script()
@@ -21,7 +21,7 @@ function getServer (port) {
   return server
 }
 
-async function start (server, opts) {
+async function start (server, opts, subOpts) {
   opts = opts || {}
 
   const plugin = {
@@ -34,7 +34,7 @@ async function start (server, opts) {
 
   await server.register([Nes, plugin])
 
-  server.subscriptionFar('/echo')
+  server.subscriptionFar('/echo', subOpts)
   server.route({
     path: '/echo',
     method: 'POST',
@@ -185,6 +185,25 @@ experiment('nes work as normal', async () => {
   pubSubTest()
 })
 
+experiment('with subscription filter', async () => {
+  let server
+
+  beforeEach(async () => {
+    server = await start(getServer(), null, {
+      filter: function (path, message, options) {
+        return true
+      }
+    })
+  })
+
+  afterEach(async () => {
+    await server.stop()
+    server = null
+  })
+
+  pubSubTest()
+})
+
 experiment('with shared mqemitter', () => {
   let server1
   let server2