Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 31 additions & 13 deletions lib/pubsub/index.js
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,19 @@
var conn = require('../common/connection.js'),
util = require('../common/util.js');

var PUBSUB_BASE_URL = 'https://www.googleapis.com/pubsub/v1beta1',
SCOPES = [
'https://www.googleapis.com/auth/pubsub',
'https://www.googleapis.com/auth/cloud-platform'
];
/**
* Base URL for Pub/Sub API.
* @type {String}
*/
var PUBSUB_BASE_URL = 'https://www.googleapis.com/pubsub/v1beta1';
/**
* Required scopes for Pub/Sub API.
* @type {Array}
*/
var SCOPES = [
'https://www.googleapis.com/auth/pubsub',
'https://www.googleapis.com/auth/cloud-platform'
];

// TODO(jbd): Emit message and error events if in polled-mode.
// sub.on('meessage', console.log)
Expand All @@ -48,21 +56,31 @@ Subscription.prototype.ack = function(ids, callback) {

/**
* Pulls from the subscribed topic.
* @param {Boolean} returnImmediately If set, the system will respond immediately.
* Otherwise, wait until new messages are
* available. Returns if timeout is reached.
* @param {Boolean} autoAck Automatically acknowledges the
* backend after pulling. By default, true.
* @param {Function} callback Callback function.
* @param {Boolean} opts.returnImmediately If set, the system will respond immediately.
* Otherwise, wait until new messages are
* available. Returns if timeout is reached.
* @param {Boolean} opts.autoAck Automatically acknowledges the
* message once it's pulled.
* @param {Function} callback Callback function.
*/
Subscription.prototype.pull = function(opts, callback) {
// TODO(jbd): Make opts optional.
var that = this;
var autoAck = !!opts.autoAck;
var body = {
subscription: this.name,
returnImmediately: !!opts.returnImmediately
};
// TODO(jbd): Auto acknowledge.
this.conn.makeReq('POST', 'subscriptions/pull', null, body, callback);
this.conn.makeReq('POST', 'subscriptions/pull', null, body, function(err, message) {
if (err) { return callback(err); }
if (!autoAck) {
return callback(null, message);
}
that.ack(message.ackId, function(err) {
if (err) { return callback(err); }
callback(null, message);
});
});
};

/**
Expand Down