Merge pull request #7 from piercus/develop

Develop

Author: piercus

lib/task.js

@@ -54,6 +54,7 @@ Task.prototype.report = function () {
 
 Task.prototype.succeed = function (res) {
 	this.worker.succeed({
+		input: this.input,
 		output: res,
 		taskToken: this.taskToken,
 		workerName: this.workerName
@@ -64,6 +65,7 @@ Task.prototype.succeed = function (res) {
 Task.prototype.fail = function (err) {
 	this.worker.fail({
 		error: err,
+		input: this.input,
 		taskToken: this.taskToken,
 		workerName: this.workerName
 	});
@@ -72,6 +74,7 @@ Task.prototype.fail = function (err) {
 
 Task.prototype.heartbeat = function () {
 	this.worker.heartbeat({
+		input: this.input,
 		taskToken: this.taskToken,
 		workerName: this.workerName
 	});

lib/worker.js

@@ -1,6 +1,7 @@
 const EventEmitter = require('events').EventEmitter;
 const util = require('util');
 const AWS = require('aws-sdk');
+const parser = require('aws-arn-parser');
 
 const Pooler = require('./pooler.js');
 const replaceError = require('./replace-error.js');
@@ -28,8 +29,7 @@ function Worker(options) {
 	this.autoStart = typeof (options.autoStart) === 'boolean' ? options.autoStart : true;
 
 	if (!options.activityArn) {
-		this.emit('error', new Error('activityArn is mandatory inside Worker'));
-		return;
+		throw (new Error('activityArn is mandatory inside Worker'));
 	}
 
 	this.concurrency = typeof (options.concurrency) === 'number' ? options.concurrency : 1;
@@ -49,6 +49,12 @@ function Worker(options) {
 		throw (new TypeError('worker does not define any function'));
 	}
 
+	const {region} = parser(options.activityArn);
+
+	if (typeof (region) === 'string' && (this.stepfunction.config.region !== region)) {
+		throw (new Error(`activity ARN region (${region}) should match with AWS Region (${this.stepfunction.config.region})`));
+	}
+
 	if (this.autoStart) {
 		setImmediate(() => {
 			this.start(() => {
@@ -131,9 +137,10 @@ Worker.prototype.execute = function (input, cb, heartbeat) {
 Worker.prototype.succeed = function (res) {
 	const params = Object.assign({}, res, {output: JSON.stringify(res.output)});
 	delete params.workerName;
+	delete params.input;
 	this.stepfunction.sendTaskSuccess(params, err => {
 		if (err) {
-			this.emit('error', err);
+			this.emit('error', {err, input: res.input});
 		} else {
 			this.emit('success', res);
 		}
@@ -150,10 +157,11 @@ Worker.prototype.fail = function (res) {
 	}
 	const params = Object.assign({}, res, {error});
 	delete params.workerName;
+	delete params.input;
 	this.logger.debug('sendTaskFailure', res.error);
 	this.stepfunction.sendTaskFailure(params, err => {
 		if (err) {
-			this.emit('error', err);
+			this.emit('error', {err, input: res.input});
 		} else {
 			this.emit('failure', res);
 		}
@@ -163,11 +171,12 @@ Worker.prototype.fail = function (res) {
 Worker.prototype.heartbeat = function (res) {
 	const params = Object.assign({}, res);
 	delete params.workerName;
+	delete params.input;
 	this.logger.debug('sendTaskHeartbeat');
 
 	this.stepfunction.sendTaskHeartbeat(params, err => {
 		if (err) {
-			this.emit('error', err);
+			this.emit('error', {err, input: res.input});
 		} else {
 			this.emit('heartbeat', res);
 		}

package.json

@@ -31,13 +31,12 @@
   "homepage": "https://github.com/piercus/step-function-worker#readme",
   "devDependencies": {
     "ava": "^0.25.0",
-    "bluebird": "^3.5.0",
     "semantic-release": "^15.1.7",
-    "vows": "^0.8.1",
     "winston": "^2.4.1",
     "xo": "^0.18.2"
   },
   "dependencies": {
+    "aws-arn-parser": "^1.0.0",
     "aws-sdk": "^2.82.0"
   }
 }

test/scenarios/failed-fn.js

@@ -1,14 +1,12 @@
 const test = require('ava').test;
 const AWS = require('aws-sdk');
-const PromiseBlue = require('bluebird');
 const winston = require('winston');
 
 const StepFunctionWorker = require('../../index.js');
 const createActivity = require('../utils/create-activity');
 const cleanUp = require('../utils/clean-up');
 
-const stepfunction = new AWS.StepFunctions();
-const stepFunctionPromises = PromiseBlue.promisifyAll(stepfunction);
+const stepFunction = new AWS.StepFunctions();
 
 const logger = new winston.Logger({
 	transports: [new winston.transports.Console({
@@ -83,7 +81,7 @@ test.serial('Step function Activity Worker with A failing worker', t => {
 		});
 
 		worker.on('success', reject);
-		stepFunctionPromises.startExecutionAsync(params);
+		stepFunction.startExecution(params).promise();
 	});
 });
 

test/scenarios/test.js

@@ -1,12 +1,10 @@
 const test = require('ava').test;
-const PromiseBlue = require('bluebird');
 const AWS = require('aws-sdk');
 const StepFunctionWorker = require('../../index.js');
 const createActivity = require('../utils/create-activity');
 const cleanUp = require('../utils/clean-up');
 
-const stepfunction = new AWS.StepFunctions();
-const stepFunctionPromises = PromiseBlue.promisifyAll(stepfunction);
+const stepFunction = new AWS.StepFunctions();
 const workerName = 'test worker name';
 const stateMachineName = 'test-state-machine-' + Math.floor(Math.random() * 1000);
 const activityName = 'test-step-function-worker-' + Math.floor(Math.random() * 1000);
@@ -89,10 +87,10 @@ test.serial('Step function Activity Worker with 2 consecutive tasks', t => {
 				});
 			});
 
-			stepFunctionPromises.startExecutionAsync(params);
+			stepFunction.startExecution(params).promise();
 		});
 
-		stepFunctionPromises.startExecutionAsync(params);
+		stepFunction.startExecution(params).promise();
 	});
 });
 
@@ -163,9 +161,9 @@ test.serial('Step function with 3 concurrent worker', t => {
 		worker.on('success', onSuccess);
 		worker.on('task', onTask);
 		worker.on('error', reject);
-		stepFunctionPromises.startExecutionAsync(params1);
-		stepFunctionPromises.startExecutionAsync(params2);
-		stepFunctionPromises.startExecutionAsync(params3);
+		stepFunction.startExecution(params1).promise();
+		stepFunction.startExecution(params2).promise();
+		stepFunction.startExecution(params3).promise();
 	});
 });
 

test/utils/clean-up.js

@@ -1,8 +1,6 @@
-const PromiseBlue = require('bluebird');
 const AWS = require('aws-sdk');
 
-const stepfunction = new AWS.StepFunctions();
-const stepFunctionPromises = PromiseBlue.promisifyAll(stepfunction);
+const stepFunction = new AWS.StepFunctions();
 
 module.exports = function ({
 	activityArn = null,
@@ -11,16 +9,16 @@ module.exports = function ({
 	let p1;
 	let p2;
 	if (activityArn) {
-		p1 = stepFunctionPromises.deleteActivityAsync({
+		p1 = stepFunction.deleteActivity({
 			activityArn
-		});
+		}).promise();
 	} else {
 		p1 = Promise.resolve();
 	}
 	if (stateMachineArn) {
-		p2 = stepFunctionPromises.deleteStateMachineAsync({
+		p2 = stepFunction.deleteStateMachine({
 			stateMachineArn
-		});
+		}).promise();
 	} else {
 		p2 = Promise.resolve();
 	}

test/utils/create-activity.js

@@ -1,8 +1,6 @@
-const PromiseBlue = require('bluebird');
 const AWS = require('aws-sdk');
 
-const stepfunction = new AWS.StepFunctions();
-const stepFunctionPromises = PromiseBlue.promisifyAll(stepfunction);
+const stepFunction = new AWS.StepFunctions();
 
 const stateMachineDefinition = function (options) {
 	return {
@@ -26,20 +24,22 @@ if (!stateMachineRoleArn) {
 }
 
 module.exports = function ({context = {}, activityName, workerName, stateMachineName}) {
-	return stepFunctionPromises
-		.createActivityAsync({
+	return stepFunction
+		.createActivity({
 			name: activityName
-		}).bind(context).then(data => {
+		}).promise().then(data => {
 			context.activityArn = data.activityArn;
 			context.workerName = workerName;
-		}).then(function () {
+		}).then(() => {
 			const params = {
-				definition: JSON.stringify(stateMachineDefinition({activityArn: this.activityArn})), /* Required */
+				definition: JSON.stringify(stateMachineDefinition({activityArn: context.activityArn})), /* Required */
 				name: stateMachineName, /* Required */
 				roleArn: stateMachineRoleArn /* Required */
 			};
-			return stepFunctionPromises.createStateMachineAsync(params);
+			return stepFunction.createStateMachine(params).promise();
 		}).then(data => {
 			context.stateMachineArn = data.stateMachineArn;
-		}).return(context);
+		}).then(() => {
+			return context;
+		});
 };