Make subscribe/unsubscribe/sendMessage return Promises that fire on RPC return.

This commit is contained in:
Ian Gulliver
2014-05-17 16:52:28 +03:00
parent 96b17ad6ff
commit 994afde51f
2 changed files with 81 additions and 29 deletions

View File

@@ -84,22 +84,26 @@ Cosmopolite.prototype.subscribe = function(subject, messages, keys) {
if (!this.ready_) { if (!this.ready_) {
throw "cosmopolite: not ready"; throw "cosmopolite: not ready";
} }
keys = keys || [];
if (subject in this.subscriptions_) { return new Promise(function(resolve, reject) {
console.log( keys = keys || [];
this.loggingPrefix_(), if (subject in this.subscriptions_) {
'not sending duplication subscription request for subject:', subject); console.log(
return; this.loggingPrefix_(),
} 'not sending duplication subscription request for subject:', subject);
this.subscriptions_[subject] = { resolve();
'messages': [], }
'keys': {}, this.subscriptions_[subject] = {
}; 'messages': [],
this.sendRPC_('subscribe', { 'keys': {},
'subject': subject, };
'messages': messages, var args = {
'keys': keys, 'subject': subject,
}); 'messages': messages,
'keys': keys,
};
this.sendRPC_('subscribe', args, resolve);
}.bind(this));
}; };
/** /**
@@ -114,10 +118,14 @@ Cosmopolite.prototype.unsubscribe = function(subject) {
if (!this.ready_) { if (!this.ready_) {
throw "cosmopolite: not ready"; throw "cosmopolite: not ready";
} }
delete this.subscriptions_[subject];
this.sendRPC_('unsubscribe', { return new Promise(function(resolve, reject) {
'subject': subject, delete this.subscriptions_[subject];
}); var args = {
'subject': subject,
}
this.sendRPC_('unsubscribe', args, resolve);
}.bind(this));
}; };
/** /**
@@ -131,15 +139,18 @@ Cosmopolite.prototype.sendMessage = function(subject, message, key) {
if (!this.ready_) { if (!this.ready_) {
throw "cosmopolite: not ready"; throw "cosmopolite: not ready";
} }
var args = {
'subject': subject, return new Promise(function(resolve, reject) {
'message': JSON.stringify(message), var args = {
'sender_message_id': this.uuid_(), 'subject': subject,
}; 'message': JSON.stringify(message),
if (key) { 'sender_message_id': this.uuid_(),
args['key'] = key; };
} if (key) {
this.sendRPC_('sendMessage', args); args['key'] = key;
}
this.sendRPC_('sendMessage', args, resolve);
}.bind(this));
}; };
/** /**

View File

@@ -174,6 +174,47 @@ asyncTest('Complex object', function() {
var cosmo = new Cosmopolite(callbacks, null, randstring()); var cosmo = new Cosmopolite(callbacks, null, randstring());
}); });
asyncTest('sendMessage Promise', function() {
expect(1);
var subject = randstring();
var message = randstring();
var callbacks = {
'onReady': function() {
cosmo.sendMessage(subject, message).then(function() {
ok(true, 'sendMessage Promise fulfilled');
cosmo.shutdown();
start();
});
},
};
var cosmo = new Cosmopolite(callbacks, null, randstring());
});
asyncTest('subscribe/unsubscribe Promise', function() {
expect(2);
var subject = randstring();
var message = randstring();
var callbacks = {
'onReady': function() {
cosmo.subscribe(subject).then(function() {
ok(true, 'subscribe Promise fulfilled');
cosmo.unsubscribe(subject).then(function() {
ok(true, 'unsubscribe Promise fulfilled');
cosmo.shutdown();
start();
});
});
},
};
var cosmo = new Cosmopolite(callbacks, null, randstring());
});
module('dev_appserver only'); module('dev_appserver only');