Add messageBuffer event to handle binary message

README.md

@@ -98,6 +98,12 @@ redis.on('message', function (channel, message) {
   // Receive message Hello again! from channel music
   console.log('Receive message %s from channel %s', message, channel);
 });
+
+// There's also a event called 'messageBuffer', which is same to 'message' except
+// return buffers instead of strings.
+redis.on('messageBuffer', function (channel, message) {
+  // Both `channel` and `message` are buffers.
+});
 ```
 When a client issues a SUBSCRIBE or PSUBSCRIBE, that connection is put into a "subscriber" mode.
 At that point, only commands that modify the subscription set are valid.

lib/redis/mixin/prototype/parser.js

@@ -25,7 +25,10 @@ exports._initParser = function () {
     }
   });
   this.replyParser.on('reply', function (reply) {
-    self.returnReply(reply);
+    // Prevent the exception be caught by the parser.
+    process.nextTick(function () {
+      self.returnReply(reply);
+    });
   });
   // "error" is bad.  Somehow the parser got confused.  It'll try to reset and continue.
   this.replyParser.on('error', function (err) {
@@ -83,10 +86,21 @@ exports.returnReply = function (reply) {
     var replyType = Array.isArray(reply) ? reply[0].toString() : null;
     switch (replyType) {
       case 'message':
-        this.emit('message', reply[1].toString(), reply[2]); // channel, message
+        if (this.listeners('message').length > 0) {
+          this.emit('message', reply[1].toString(), reply[2].toString());
+        }
+        if (this.listeners('messageBuffer').length > 0) {
+          this.emit('messageBuffer', reply[1], reply[2]);
+        }
         break;
       case 'pmessage':
-        this.emit('pmessage', reply[1].toString(), reply[2].toString(), reply[3]); // pattern, channel, message
+        var pattern = reply[1].toString();
+        if (this.listeners('pmessage').length > 0) {
+          this.emit('pmessage', pattern, reply[2].toString(), reply[3].toString());
+        }
+        if (this.listeners('pmessageBuffer').length > 0) {
+          this.emit('pmessageBuffer', pattern, reply[2], reply[3]);
+        }
         break;
       case 'subscribe':
       case 'psubscribe':

test/functional/pub_sub.js

@@ -46,6 +46,58 @@ describe('pub/sub', function () {
     });
   });
 
+  it('should receive messages when subscribe a channel', function (done) {
+    var redis = new Redis();
+    var pub = new Redis();
+    var pending = 2;
+    redis.subscribe('foo', function () {
+      pub.publish('foo', 'bar');
+    });
+    redis.on('message', function (channel, message) {
+      expect(channel).to.eql('foo');
+      expect(message).to.eql('bar');
+      if (!--pending) {
+        done();
+      }
+    });
+    redis.on('messageBuffer', function (channel, message) {
+      expect(channel).to.be.instanceof(Buffer);
+      expect(channel.toString()).to.eql('foo');
+      expect(message).to.be.instanceof(Buffer);
+      expect(message.toString()).to.eql('bar');
+      if (!--pending) {
+        done();
+      }
+    });
+  });
+
+  it('should receive messages when psubscribe a pattern', function (done) {
+    var redis = new Redis();
+    var pub = new Redis();
+    var pending = 2;
+    redis.psubscribe('f?oo', function () {
+      pub.publish('fzoo', 'bar');
+    });
+    redis.on('pmessage', function (pattern, channel, message) {
+      expect(pattern).to.eql('f?oo');
+      expect(channel).to.eql('fzoo');
+      expect(message).to.eql('bar');
+      if (!--pending) {
+        done();
+      }
+    });
+    redis.on('pmessageBuffer', function (pattern, channel, message) {
+      expect(pattern).to.eql('f?oo');
+      expect(channel).to.be.instanceof(Buffer);
+      expect(channel.toString()).to.eql('fzoo');
+      expect(message).to.be.instanceof(Buffer);
+      expect(message.toString()).to.eql('bar');
+      if (!--pending) {
+        done();
+      }
+    });
+  });
+
   it('should exit subscriber mode using punsubscribe', function (done) {
     var redis = new Redis();
     redis.psubscribe('f?oo', 'b?ar', function () {