Merge pull request #13 from piercus/restart-activity-worker

Restart activity worker

Author: piercus

lib/pooler.js

@@ -1,5 +1,5 @@
 const util = require('util');
-const EventEmitter = require('events').EventEmitter;
+const {EventEmitter} = require('events');
 const Task = require('./task.js');
 
 /**
@@ -34,8 +34,8 @@ Pooler.prototype.stop = function (cb) {
 			this.removeAllListeners();
 			cb();
 		});
-    // This would be better approach but it does not seem to work
-    // this._request.abort();
+		// This would be better approach but it does not seem to work
+		// this._request.abort();
 	} else {
 		cb();
 	}
@@ -85,9 +85,9 @@ Pooler.prototype.getActivityTask = function () {
 	}, (err, data) => {
 		this._request = null;
 		if (err) {
-      // Console.log(err);
+			// Console.log(err);
 			if (err.code === 'RequestAbortedError') {
-        // In case of abort, close silently
+				// In case of abort, close silently
 			} else {
 				this.emit('error', err);
 			}

lib/task.js

@@ -1,4 +1,4 @@
-const EventEmitter = require('events').EventEmitter;
+const {EventEmitter} = require('events');
 const util = require('util');
 
 /**

lib/worker.js

@@ -1,4 +1,4 @@
-const EventEmitter = require('events').EventEmitter;
+const {EventEmitter} = require('events');
 const util = require('util');
 const AWS = require('aws-sdk');
 const parser = require('aws-arn-parser');
@@ -69,7 +69,10 @@ function Worker(options) {
 * @param {function} callback
 */
 Worker.prototype.start = function (cb) {
-	this.updatePool(cb);
+	this.updatePool(err => {
+		this.logger.info('Worker started');
+		cb(err);
+	});
 };
 /**
 * Get a report of the actual situation of the worker
@@ -117,25 +120,48 @@ Worker.prototype.removePooler = function (cb) {
 
 /**
 * Close the worker, this function might take 60 seconds to finish to do step function design
+* remove all the events attached to the worker
 * @param {function} callback
 */
 
 Worker.prototype.close = function (cb) {
-	this.logger.info('Closing the worker ... this might take 60 seconds');
+	this.stop(cb);
+	this.removeAllListeners();
+};
+
+/**
+* Stop the worker
+* But does not remove all the events attached to it
+* NB: worker.concurrency is set to 0
+* @param {function} callback
+*/
+
+Worker.prototype.stop = function (cb) {
+	this.logger.info('Stopping the worker ... this might take 60 seconds');
 	this.concurrency = 0;
 	this.updatePool(err => {
-		this.logger.info('Worker closed');
+		this.logger.info('Worker stopped');
 		cb(err);
 	});
-	this.removeAllListeners();
+};
+
+Worker.prototype.restart = function (cb) {
+	const oldConcurrency = this.concurrency;
+	this.stop(err => {
+		if (err) {
+			return cb(err);
+		}
+		this.concurrency = oldConcurrency;
+		this.start(cb);
+	});
 };
 
 Worker.prototype.execute = function (input, cb, heartbeat) {
 	setImmediate(() => {
 		try {
 			this.fn(input, cb, heartbeat);
-		} catch (err) {
-			cb(err);
+		} catch (error) {
+			cb(error);
 		}
 	});
 };