BREAKING CHANGES:

Redesign of the whole architecture

Author: piercus

lib/worker.js

@@ -209,6 +209,21 @@ Worker.prototype.removePooler = function (pooler) {
 	}
 };
 
+Worker.prototype.removeTask = function (pooler) {
+	this.logger.debug('removePooler');
+
+	const index = this._poolers.indexOf(pooler);
+	if (index === -1) {
+		throw (new Error(`pooler ${pooler} is not in the pooler list`));
+	}
+
+	this._poolers.splice(index, 1);
+
+	if (this._poolers.length === 0) {
+		this.emit('empty-poolers');
+	}
+};
+
 // Worker.prototype.removePooler = function () {
 // 	if(!this._poolerRemovalPromise){
 // 		this._poolerRemovalPromise = Promise.resolve()
@@ -290,10 +305,16 @@ Worker.prototype.stop = function () {
 
 Worker.prototype.restart = function (cb) {
 	const oldPoolConcurrency = this.poolConcurrency;
-	return this.stop().then(() => {
+
+	const promise = this.stop().then(() => {
 		this.poolConcurrency = oldPoolConcurrency;
-		return this.start(cb);
+		return this.start();
 	});
+	if (cb) {
+		promise.catch(cb).then(() => cb());
+	} else {
+		return promise;
+	}
 };
 
 util.inherits(Worker, EventEmitter);

test/scenarios/issue-16.js

@@ -131,6 +131,7 @@ test.serial('Step function Activity Worker with 200 parallel tasks and heartbeat
 	return new Promise((resolve, reject) => {
 		worker.once('empty', () => {
 			t.is(count, totalTasks);
+			t.true(countFull > 0);
 			// T.is(Math.abs(countFull - (totalTasks-taskConcurrency))/totalTasks)
 			const endDate = new Date();
 			worker.logger.info(`Spent ${(endDate - startDate) / 1000} seconds`);

test/scenarios/test.js

@@ -197,7 +197,9 @@ test.serial('Restart the worker', t => {
 
 			if (countSuccess === 1) {
 				const beforeRestartLength = worker._poolers.length;
+				console.log('restart');
 				worker.restart(() => {
+					console.log('restarted');
 					t.is(worker._poolers.length, beforeRestartLength);
 					stepFunction.startExecution(params2).promise();
 				});