refact(rpc): pubsub improvements

Author: artch

lib/rpc.js

@@ -58,14 +58,29 @@ class RpcServer {
         this.socket = socket;
         this.socket.pipe(new JSONFrameStream(this._processFrame.bind(this)));
         this.methods = methods;
+        this.channelUnsubscribe = new Map();
+        this.socket.on('close', () => {
+            this.channelUnsubscribe.forEach(unsubscribe => unsubscribe());
+            this.channelUnsubscribe.clear();
+        });
     }
 
     _processFrame(obj) {
         let args = obj.args || [];
         if(obj.method == 'subscribe') {
-            this.methods.subscribe(obj.channel, (pubsub) => {
-                this.socket.write(JSONFrameStream.makeFrame({pubsub}));
-            });
+            if(this.channelUnsubscribe.has('*')) {
+                return;
+            }
+            if(obj.channel === '*') {
+                this.channelUnsubscribe.forEach(unsubscribe => unsubscribe());
+                this.channelUnsubscribe.clear();
+            }
+            if(!this.channelUnsubscribe.has(obj.channel)) {
+                let unsubscribe = this.methods.subscribe(obj.channel, (pubsub) => {
+                    this.socket.write(JSONFrameStream.makeFrame({pubsub}));
+                });
+                this.channelUnsubscribe.set(obj.channel, unsubscribe);
+            }
             return;
         }
         this.methods[obj.method].apply(null, args.concat([(error, result) => {
@@ -119,13 +134,15 @@ class RpcClient {
         this.defers.set(this.requestId, defer);
         return defer.promise;
     }
-    subscribe(channel, cb) {
+    subscribe(channelToSubscribe, callback) {
         let request = {
             method: 'subscribe',
-            channel
+            channel: channelToSubscribe
         };
         this.socket.write(JSONFrameStream.makeFrame(request));
-        this.pubsub.addListener(channel, cb);
+        this.pubsub.addListener(channelToSubscribe, (channel, ...args) => {
+            callback.apply({channel}, args);
+        });
     }
 }