co: stop using co.wrap.

This commit is contained in:
Christopher Jeffrey 2017-01-05 04:30:29 -08:00
parent cca763ca95
commit 94d4e80373
No known key found for this signature in database
GPG Key ID: 8962AB9DE6666BBD
6 changed files with 128 additions and 67 deletions

View File

@ -65,7 +65,7 @@ function BIP150(bip151, hostname, outbound, db, identity) {
this.challengeSent = false; this.challengeSent = false;
this.auth = false; this.auth = false;
this.completed = false; this.completed = false;
this.callback = null; this.job = null;
this.timeout = null; this.timeout = null;
} }
@ -235,44 +235,56 @@ BIP150.prototype.destroy = function destroy() {
} }
}; };
BIP150.prototype.complete = function complete(err) { BIP150.prototype.cleanup = function cleanup(err) {
var job = this.job;
assert(!this.completed, 'Already completed.'); assert(!this.completed, 'Already completed.');
assert(this.callback, 'No completion callback.'); assert(job, 'No completion job.');
this.completed = true; this.completed = true;
this.job = null;
if (this.timeout != null) { if (this.timeout != null) {
clearTimeout(this.timeout); clearTimeout(this.timeout);
this.timeout = null; this.timeout = null;
} }
this.callback(err); return job;
this.callback = null; };
BIP150.prototype.resolve = function resolve(result) {
var job = this.cleanup();
job.resolve(result);
};
BIP150.prototype.reject = function reject(err) {
var job = this.cleanup();
job.reject(err);
}; };
BIP150.prototype.wait = function wait(timeout) { BIP150.prototype.wait = function wait(timeout) {
var self = this; var self = this;
return new Promise(function(resolve, reject) { return new Promise(function(resolve, reject) {
self._wait(timeout, co.wrap(resolve, reject)); self._wait(timeout, resolve, reject);
}); });
}; };
BIP150.prototype._wait = function wait(timeout, callback) { BIP150.prototype._wait = function wait(timeout, resolve, reject) {
var self = this; var self = this;
assert(!this.auth, 'Cannot wait for init after handshake.'); assert(!this.auth, 'Cannot wait for init after handshake.');
this.callback = callback; this.job = co.job(resolve, reject);
if (this.outbound && !this.peerIdentity) if (this.outbound && !this.peerIdentity)
return this.complete(new Error('No identity for ' + this.hostname + '.')); return this.reject(new Error('No identity for ' + this.hostname + '.'));
this.timeout = setTimeout(function() { this.timeout = setTimeout(function() {
self.complete(new Error('BIP150 handshake timed out.')); self.reject(new Error('BIP150 handshake timed out.'));
}, timeout); }, timeout);
this.once('auth', function() { this.once('auth', function() {
self.complete(); self.resolve();
}); });
}; };

View File

@ -292,7 +292,7 @@ BIP151Stream.prototype.verify = function verify(tag) {
* @property {Boolean} initSent * @property {Boolean} initSent
* @property {Boolean} ackSent * @property {Boolean} ackSent
* @property {Object} timeout * @property {Object} timeout
* @property {Function} callback * @property {Job} job
* @property {Boolean} completed * @property {Boolean} completed
* @property {Boolean} handshake * @property {Boolean} handshake
*/ */
@ -311,7 +311,7 @@ function BIP151(cipher) {
this.initSent = false; this.initSent = false;
this.ackSent = false; this.ackSent = false;
this.timeout = null; this.timeout = null;
this.callback = null; this.job = null;
this.completed = false; this.completed = false;
this.handshake = false; this.handshake = false;
@ -435,22 +435,43 @@ BIP151.prototype.encack = function encack(publicKey) {
}; };
/** /**
* Complete the timeout for handshake, * Cleanup handshake job.
* possibly with an error. * @returns {Job}
* @param {Error?} err
*/ */
BIP151.prototype.complete = function complete(err) { BIP151.prototype.cleanup = function cleanup() {
var job = this.job;
assert(!this.completed, 'Already completed.'); assert(!this.completed, 'Already completed.');
assert(this.callback, 'No completion callback.'); assert(job, 'No completion job.');
this.completed = true; this.completed = true;
this.job = null;
clearTimeout(this.timeout); clearTimeout(this.timeout);
this.timeout = null; this.timeout = null;
this.callback(err); return job;
this.callback = null; };
/**
* Complete the timeout for handshake.
* @param {Object} result
*/
BIP151.prototype.resolve = function resolve(result) {
var job = this.cleanup();
job.resolve(result);
};
/**
* Complete the timeout for handshake with error.
* @param {Error} err
*/
BIP151.prototype.reject = function reject(err) {
var job = this.cleanup();
job.reject(err);
}; };
/** /**
@ -462,28 +483,31 @@ BIP151.prototype.complete = function complete(err) {
BIP151.prototype.wait = function wait(timeout) { BIP151.prototype.wait = function wait(timeout) {
var self = this; var self = this;
return new Promise(function(resolve, reject) { return new Promise(function(resolve, reject) {
self._wait(timeout, co.wrap(resolve, reject)); self._wait(timeout, resolve, reject);
}); });
}; };
/** /**
* Set a timeout and wait for handshake to complete. * Set a timeout and wait for handshake to complete.
* @private * @private
* @param {Number} timeout
* @param {Function} resolve
* @param {Function} reject
*/ */
BIP151.prototype._wait = function wait(timeout, callback) { BIP151.prototype._wait = function wait(timeout, resolve, reject) {
var self = this; var self = this;
assert(!this.handshake, 'Cannot wait for init after handshake.'); assert(!this.handshake, 'Cannot wait for init after handshake.');
this.callback = callback; this.job = co.job(resolve, reject);
this.timeout = setTimeout(function() { this.timeout = setTimeout(function() {
self.complete(new Error('BIP151 handshake timed out.')); self.reject(new Error('BIP151 handshake timed out.'));
}, timeout); }, timeout);
this.once('handshake', function() { this.once('handshake', function() {
self.complete(); self.resolve();
}); });
}; };

View File

@ -2977,7 +2977,7 @@ function RequestEntry() {
} }
RequestEntry.prototype.addJob = function addJob(resolve, reject) { RequestEntry.prototype.addJob = function addJob(resolve, reject) {
this.jobs.push(new Job(resolve, reject)); this.jobs.push(co.job(resolve, reject));
}; };
RequestEntry.prototype.setTimeout = function setTimeout(timeout) { RequestEntry.prototype.setTimeout = function setTimeout(timeout) {
@ -3006,16 +3006,6 @@ RequestEntry.prototype.resolve = function resolve(result) {
this.jobs.length = 0; this.jobs.length = 0;
}; };
/**
* Job
* @constructor
*/
function Job(resolve, reject) {
this.resolve = resolve;
this.reject = reject;
}
/* /*
* Expose * Expose
*/ */

View File

@ -1158,7 +1158,7 @@ Pool.prototype._handleHeaders = co(function* handleHeaders(peer, headers) {
* @returns {Promise} * @returns {Promise}
*/ */
Pool.prototype.handleBlockInv = co(function* handleBlockInv(hashes, peer) { Pool.prototype.handleBlockInv = co(function* handleBlockInv(peer, hashes) {
var unlock = yield this.locker.lock(); var unlock = yield this.locker.lock();
try { try {
return yield this._handleBlockInv(peer, hashes); return yield this._handleBlockInv(peer, hashes);
@ -2135,7 +2135,7 @@ util.inherits(BroadcastItem, EventEmitter);
*/ */
BroadcastItem.prototype.addJob = function addJob(resolve, reject) { BroadcastItem.prototype.addJob = function addJob(resolve, reject) {
this.jobs.push(new Job(resolve, reject)); this.jobs.push(co.job(resolve, reject));
}; };
/** /**
@ -2289,16 +2289,6 @@ BroadcastItem.prototype.inspect = function inspect() {
+ '>'; + '>';
}; };
/**
* Job
* @constructor
*/
function Job(resolve, reject) {
this.resolve = resolve;
this.reject = reject;
}
/* /*
* Expose * Expose
*/ */

View File

@ -293,6 +293,29 @@ every = co(function* every(jobs) {
return true; return true;
}); });
/**
* Create a job object.
* @returns {Job}
*/
function job(resolve, reject) {
return new Job(resolve, reject);
}
/**
* Job
* @constructor
* @param {Function} resolve
* @param {Function} reject
* @property {Function} resolve
* @property {Function} reject
*/
function Job(resolve, reject) {
this.resolve = resolve;
this.reject = reject;
}
/* /*
* This drives me nuts. * This drives me nuts.
*/ */
@ -324,5 +347,6 @@ exports.wrap = wrap;
exports.call = call; exports.call = call;
exports.promisify = promisify; exports.promisify = promisify;
exports.every = every; exports.every = every;
exports.job = job;
module.exports = exports; module.exports = exports;

View File

@ -662,7 +662,7 @@ Worker.prototype.destroy = function destroy() {
Worker.prototype.execute = function execute(packet, timeout) { Worker.prototype.execute = function execute(packet, timeout) {
var self = this; var self = this;
return new Promise(function(resolve, reject) { return new Promise(function(resolve, reject) {
self._execute(packet, timeout, co.wrap(resolve, reject)); self._execute(packet, timeout, resolve, reject);
}); });
}; };
@ -671,12 +671,13 @@ Worker.prototype.execute = function execute(packet, timeout) {
* @private * @private
* @param {Packet} packet * @param {Packet} packet
* @param {Number} timeout * @param {Number} timeout
* @param {Function} callback * @param {Function} resolve
* @param {Function} reject
* the worker method specifies. * the worker method specifies.
*/ */
Worker.prototype._execute = function _execute(packet, timeout, callback) { Worker.prototype._execute = function _execute(packet, timeout, resolve, reject) {
var job = new PendingJob(this, packet.id, callback); var job = new PendingJob(this, packet.id, resolve, reject);
assert(!this.pending[packet.id], 'ID overflow.'); assert(!this.pending[packet.id], 'ID overflow.');
@ -699,7 +700,7 @@ Worker.prototype.resolveJob = function resolveJob(id, result) {
if (!job) if (!job)
throw new Error('Job ' + id + ' is not in progress.'); throw new Error('Job ' + id + ' is not in progress.');
job.finish(null, result); job.resolve(result);
}; };
/** /**
@ -714,7 +715,7 @@ Worker.prototype.rejectJob = function rejectJob(id, err) {
if (!job) if (!job)
throw new Error('Job ' + id + ' is not in progress.'); throw new Error('Job ' + id + ' is not in progress.');
job.finish(err); job.reject(err);
}; };
/** /**
@ -737,13 +738,14 @@ Worker.prototype.killJobs = function killJobs() {
* @constructor * @constructor
* @param {Worker} worker * @param {Worker} worker
* @param {Number} id * @param {Number} id
* @param {Function} callback * @param {Function} resolve
* @param {Function} reject
*/ */
function PendingJob(worker, id, callback) { function PendingJob(worker, id, resolve, reject) {
this.worker = worker; this.worker = worker;
this.id = id; this.id = id;
this.callback = callback; this.job = co.job(resolve, reject);
this.timer = null; this.timer = null;
} }
@ -759,7 +761,7 @@ PendingJob.prototype.start = function start(timeout) {
return; return;
this.timer = setTimeout(function() { this.timer = setTimeout(function() {
self.finish(new Error('Worker timed out.')); self.reject(new Error('Worker timed out.'));
}, timeout); }, timeout);
}; };
@ -768,21 +770,20 @@ PendingJob.prototype.start = function start(timeout) {
*/ */
PendingJob.prototype.destroy = function destroy() { PendingJob.prototype.destroy = function destroy() {
this.finish(new Error('Job was destroyed.')); this.reject(new Error('Job was destroyed.'));
}; };
/** /**
* Complete job with error and result. * Cleanup job state.
* @param {Error} err * @returns {Job}
* @param {Object} reject
*/ */
PendingJob.prototype.finish = function finish(err, result) { PendingJob.prototype.cleanup = function cleanup() {
var callback = this.callback; var job = this.job;
assert(callback, 'Already finished.'); assert(job, 'Already finished.');
this.callback = null; this.job = null;
if (this.timer != null) { if (this.timer != null) {
clearTimeout(this.timer); clearTimeout(this.timer);
@ -792,7 +793,27 @@ PendingJob.prototype.finish = function finish(err, result) {
assert(this.worker.pending[this.id]); assert(this.worker.pending[this.id]);
delete this.worker.pending[this.id]; delete this.worker.pending[this.id];
callback(err, result); return job;
};
/**
* Complete job with result.
* @param {Object} result
*/
PendingJob.prototype.resolve = function resolve(result) {
var job = this.cleanup();
job.resolve(result);
};
/**
* Complete job with error.
* @param {Error} err
*/
PendingJob.prototype.reject = function reject(err) {
var job = this.cleanup();
job.reject(err);
}; };
/* /*