DEV: added unsubscribeAll method

Author: Isan-Rivkin

src/pubsub.js

@@ -19,6 +19,67 @@ module.exports = (arg) => {
   const ps = new EventEmitter()
   const subscriptions = {}
   ps.id = Math.random()
+  // unsubscribe from topics 
+  const _internalUnsubscribe = (input)=>{
+    let {topic,handler,callback} = input
+    if (!isNode) {
+      if (!callback) {
+        return Promise.reject(NotSupportedError())
+      }
+
+      return setImmediate(() => callback(NotSupportedError()))
+    }
+
+    if (ps.listenerCount(topic) === 0 || !subscriptions[topic]) {
+      const err = new Error(`Not subscribed to '${topic}'`)
+
+      if (!callback) {
+        return Promise.reject(err)
+      }
+
+      return setImmediate(() => callback(err))
+    }
+    
+    // unsubscribe from specific or all handlers 
+    if(handler){
+      ps.removeListener(topic, handler)
+    }else{
+      ps.removeAllListeners(topic)
+    }
+
+    // Drop the request once we are actually done
+    if (ps.listenerCount(topic) === 0) {
+      if (!callback) {
+        return new Promise((resolve, reject) => {
+          // When the response stream has ended, resolve the promise
+          eos(subscriptions[topic].res, (err) => {
+            // FIXME: Artificial timeout needed to ensure unsubscribed
+            setTimeout(() => {
+              if (err) return reject(err)
+              resolve()
+            })
+          })
+          subscriptions[topic].req.abort()
+          subscriptions[topic] = null
+        })
+      }
+
+      // When the response stream has ended, call the callback
+      eos(subscriptions[topic].res, (err) => {
+        // FIXME: Artificial timeout needed to ensure unsubscribed
+        setTimeout(() => callback(err))
+      })
+      subscriptions[topic].req.abort()
+      subscriptions[topic] = null
+      return
+    }
+
+    if (!callback) {
+      return Promise.resolve()
+    }
+
+    setImmediate(() => callback())
+  }
   return {
     subscribe: (topic, handler, options, callback) => {
       const defaultOptions = {
@@ -59,58 +120,10 @@ module.exports = (arg) => {
       subscribe(topic, handler, options, callback)
     },
     unsubscribe: (topic, handler, callback) => {
-      if (!isNode) {
-        if (!callback) {
-          return Promise.reject(NotSupportedError())
-        }
-
-        return setImmediate(() => callback(NotSupportedError()))
-      }
-
-      if (ps.listenerCount(topic) === 0 || !subscriptions[topic]) {
-        const err = new Error(`Not subscribed to '${topic}'`)
-
-        if (!callback) {
-          return Promise.reject(err)
-        }
-
-        return setImmediate(() => callback(err))
-      }
-
-      ps.removeListener(topic, handler)
-
-      // Drop the request once we are actually done
-      if (ps.listenerCount(topic) === 0) {
-        if (!callback) {
-          return new Promise((resolve, reject) => {
-            // When the response stream has ended, resolve the promise
-            eos(subscriptions[topic].res, (err) => {
-              // FIXME: Artificial timeout needed to ensure unsubscribed
-              setTimeout(() => {
-                if (err) return reject(err)
-                resolve()
-              })
-            })
-            subscriptions[topic].req.abort()
-            subscriptions[topic] = null
-          })
-        }
-
-        // When the response stream has ended, call the callback
-        eos(subscriptions[topic].res, (err) => {
-          // FIXME: Artificial timeout needed to ensure unsubscribed
-          setTimeout(() => callback(err))
-        })
-        subscriptions[topic].req.abort()
-        subscriptions[topic] = null
-        return
-      }
-
-      if (!callback) {
-        return Promise.resolve()
-      }
-
-      setImmediate(() => callback())
+      _internalUnsubscribe({topic:topic,handler:handler,callback:callback});
+    },
+    unsubscribeAll : (topic,callback)=>{
+      _internalUnsubscribe({topic : topic, callback: callback});
     },
     publish: promisify((topic, data, callback) => {
       if (!isNode) {

test/sub-modules.spec.js

@@ -160,6 +160,7 @@ describe('submodules', () => {
 
     expect(pubsub.subscribe).to.be.a('function')
     expect(pubsub.unsubscribe).to.be.a('function')
+    expect(pubsub.unsubscribeAll).to.be.a('function')
     expect(pubsub.publish).to.be.a('function')
     expect(pubsub.ls).to.be.a('function')
     expect(pubsub.peers).to.be.a('function')