pubsub: return timestamp on message (#1843)

packages/pubsub/src/index.js

@@ -568,9 +568,11 @@ PubSub.prototype.subscribe = function(topic, subName, options, callback) {
  * // Register a listener for `message` events.
  * subscription.on('message', function(message) {
  *   // Called every time a message is received.
- *   // message.id = ID used to acknowledge its receival.
+ *   // message.id = ID of the message.
+ *   // message.ackId = ID used to acknowledge the message receival.
  *   // message.data = Contents of the message.
  *   // message.attributes = Attributes of the message.
+ *   // message.timestamp = Timestamp when Pub/Sub received the message.
  * });
  */
 PubSub.prototype.subscription = function(name, options) {

packages/pubsub/src/subscription.js

@@ -126,6 +126,8 @@ var PUBSUB_API_TIMEOUT = 90000;
  *   // message.id = ID of the message.
  *   // message.ackId = ID used to acknowledge the message receival.
  *   // message.data = Contents of the message.
+ *   // message.attributes = Attributes of the message.
+ *   // message.timestamp = Timestamp when Pub/Sub received the message.
  *
  *   // Ack the message:
  *   // message.ack(callback);
@@ -332,8 +334,9 @@ function Subscription(pubsub, options) {
 modelo.inherits(Subscription, common.GrpcServiceObject, events.EventEmitter);
 
 /**
- * Simplify a message from an API response to have three properties, `id`,
- * `data` and `attributes`. `data` is always converted to a string.
+ * Simplify a message from an API response to have five properties: `id`,
+ * `ackId`, `data`, `attributes`, and `timestamp`. `data` is always converted to
+ * a string.
  *
  * @private
  */
@@ -357,6 +360,16 @@ Subscription.formatMessage_ = function(msg, encoding) {
     if (innerMessage.attributes) {
       message.attributes = innerMessage.attributes;
     }
+
+    if (innerMessage.publishTime) {
+      var publishTime = innerMessage.publishTime;
+
+      if (publishTime.seconds && publishTime.nanos) {
+        var seconds = parseInt(publishTime.seconds, 10);
+        var milliseconds = parseInt(publishTime.nanos, 10) / 1e6;
+        message.timestamp = new Date(seconds * 1000 + milliseconds);
+      }
+    }
   }
 
   return message;

packages/pubsub/src/topic.js

@@ -552,9 +552,11 @@ Topic.prototype.subscribe = function(subName, options, callback) {
  * // Register a listener for `message` events.
  * subscription.on('message', function(message) {
  *   // Called every time a message is received.
- *   // message.id = ID used to acknowledge its receival.
+ *   // message.id = ID of the message.
+ *   // message.ackId = ID used to acknowledge the message receival.
  *   // message.data = Contents of the message.
  *   // message.attributes = Attributes of the message.
+ *   // message.timestamp = Timestamp when Pub/Sub received the message.
  * });
  */
 Topic.prototype.subscription = function(name, options) {

packages/pubsub/test/subscription.js

@@ -273,21 +273,31 @@ describe('Subscription', function() {
       var obj = { hi: 'there' };
       var stringified = new Buffer(JSON.stringify(obj)).toString('base64');
       var attributes = {};
+      var publishTime = {
+        seconds: '1480413405',
+        nanos: 617000000
+      };
+
+      var seconds = parseInt(publishTime.seconds, 10);
+      var milliseconds = parseInt(publishTime.nanos, 10) / 1e6;
+      var expectedDate = new Date(seconds * 1000 + milliseconds);
 
       var msg = Subscription.formatMessage_({
         ackId: 'abc',
         message: {
           data: stringified,
           messageId: 7,
-          attributes: attributes
+          attributes: attributes,
+          publishTime: publishTime
         }
       });
 
       assert.deepEqual(msg, {
         ackId: 'abc',
         id: 7,
         data: obj,
-        attributes: attributes
+        attributes: attributes,
+        timestamp: expectedDate
       });
     });