Merge pull request #988 from roberttod/pubsub-encoding-option

Add pubsub encoding option

Author: stephenplusplus

lib/pubsub/subscription.js

@@ -273,6 +273,7 @@ function Subscription(pubsub, options) {
     is.number(options.maxInProgress) ? options.maxInProgress : Infinity;
   this.messageListeners = 0;
   this.paused = false;
+  this.encoding = is.string(options.encoding) ? options.encoding : 'utf-8';
 
   this.listenForEvents_();
 }
@@ -285,7 +286,7 @@ modelo.inherits(Subscription, ServiceObject, events.EventEmitter);
  *
  * @private
  */
-Subscription.formatMessage_ = function(msg) {
+Subscription.formatMessage_ = function(msg, encoding) {
   var innerMessage = msg.message;
   var message = {
     ackId: msg.ackId
@@ -295,7 +296,7 @@ Subscription.formatMessage_ = function(msg) {
     message.id = innerMessage.messageId;
 
     if (innerMessage.data) {
-      message.data = new Buffer(innerMessage.data, 'base64').toString('utf-8');
+      message.data = new Buffer(innerMessage.data, 'base64').toString(encoding);
 
       try {
         message.data = JSON.parse(message.data);
@@ -482,6 +483,7 @@ Subscription.prototype.delete = function(callback) {
 Subscription.prototype.pull = function(options, callback) {
   var self = this;
   var MAX_EVENTS_LIMIT = 1000;
+  var encoding = this.encoding;
 
   if (!callback) {
     callback = options;
@@ -509,7 +511,9 @@ Subscription.prototype.pull = function(options, callback) {
     }
 
     var messages = arrify(response.receivedMessages)
-      .map(Subscription.formatMessage_)
+      .map(function(msg) {
+        return Subscription.formatMessage_(msg, encoding);
+      })
       .map(self.decorateMessage_.bind(self));
 
     self.refreshPausedStatus_();

test/pubsub/subscription.js

@@ -47,6 +47,7 @@ describe('Subscription', function() {
   };
   var message = 'howdy';
   var messageBuffer = new Buffer(message).toString('base64');
+  var messageBinary = new Buffer(message).toString('binary');
   var messageObj = {
     receivedMessages: [{
       ackId: 3,
@@ -61,6 +62,11 @@ describe('Subscription', function() {
     data: message,
     id: 7
   };
+  var expectedMessageAsBinary = {
+    ackId: 3,
+    data: messageBinary,
+    id: 7
+  };
 
   before(function() {
     mockery.registerMock('../common/service-object.js', FakeServiceObject);
@@ -98,12 +104,14 @@ describe('Subscription', function() {
         name: SUB_NAME,
         autoAck: true,
         interval: 100,
-        maxInProgress: 3
+        maxInProgress: 3,
+        encoding: 'binary'
       };
       var sub = new Subscription(PUBSUB, CONFIG);
       assert.strictEqual(sub.autoAck, CONFIG.autoAck);
       assert.strictEqual(sub.interval, CONFIG.interval);
       assert.strictEqual(sub.maxInProgress, 3);
+      assert.strictEqual(sub.encoding, CONFIG.encoding);
     });
 
     it('should be closed', function() {
@@ -134,6 +142,10 @@ describe('Subscription', function() {
       assert.strictEqual(subscription.paused, false);
     });
 
+    it('should default encoding to utf-8 if not specified', function() {
+      assert.strictEqual(subscription.encoding, 'utf-8');
+    });
+
     it('should create an iam object', function() {
       assert.deepEqual(subscription.iam.calledWith_, [
         PUBSUB,
@@ -219,6 +231,12 @@ describe('Subscription', function() {
       var msg = Subscription.formatMessage_(messageObj.receivedMessages[0]);
       assert.deepEqual(msg, expectedMessage);
     });
+
+    it('should decode buffer to specified encoding', function() {
+      var msg = Subscription
+        .formatMessage_(messageObj.receivedMessages[0], 'binary');
+      assert.deepEqual(msg, expectedMessageAsBinary);
+    });
   });
 
   describe('formatName_', function() {
@@ -473,6 +491,42 @@ describe('Subscription', function() {
       subscription.pull({}, assert.ifError);
     });
 
+    describe('with encoding option', function() {
+      var formatMessageMemo, encodingMemo, onFormatMessage;
+      beforeEach(function() {
+        encodingMemo = subscription.encoding;
+        formatMessageMemo = Subscription.formatMessage_;
+        Subscription.formatMessage_ = function(msg, encoding) {
+          onFormatMessage(msg, encoding);
+          return formatMessageMemo(msg, encoding);
+        };
+      });
+      afterEach(function() {
+        Subscription.formatMessage_ = formatMessageMemo;
+        subscription.encoding = encodingMemo;
+      });
+
+      it('should call formatMessage_ with binary', function(done) {
+        onFormatMessage = function(msg, encoding) {
+          assert.equal(encoding, 'binary');
+          done();
+        };
+        subscription.encoding = 'binary';
+
+        subscription.pull({}, assert.ifError);
+      });
+
+      it('should call formatMessage_ with utf-8', function(done) {
+        onFormatMessage = function(msg, encoding) {
+          assert.equal(encoding, 'utf-8');
+          done();
+        };
+        subscription.encoding = 'utf-8';
+
+        subscription.pull({}, assert.ifError);
+      });
+    });
+
     describe('autoAck false', function() {
       beforeEach(function() {
         subscription.autoAck = false;