<NME> sqlContext.js
<BEF> const _ = require( "lodash" );
const util = require( "util" );
const EventEmitter = require( "events" );

const machina = require( "machina" );
const log = require( "debug" )( "seriate:sql" );
const addState = require( "./sqlContextUtils" ).addState;

module.exports = function() {
	const SqlContext = machina.Fsm.extend( {
		_connected( connection ) {
			this.connection = connection;
			this.handle( "success" );
		},

		_connectionError( err ) {
			this.handle( "error", err );
		},

		initialState: "uninitialized",

		initialize( options ) {
			this.results = {};
			this.pipeline = [];
			this.pipePos = -1;
			this.metrics = options.metrics;
			this.metricsNamespace = options.namespace;
			this.options = options;
			options.connection
				.then(
					this._connected.bind( this ),
					this._connectionError.bind( this )
				);
		},

		nextState() {
			this.pipePos += 1;
			const nextState = this.pipeline[ this.pipePos ] || "done";
			this.transition( nextState );
		},

		states: {
			uninitialized: {
				start: "connecting",
				"*"() {
					this.deferUntilTransition( "connecting" );
				}
			},
			connecting: {
				success() {
					this.nextState();
				},
				error( err ) {
					this.err = err;
					this.transition( "error" );
				}
			},

			done: {
				_onEnter() {
					this.emit( "end", this.results );
				}
			},

			error: {
				_onEnter() {
					const precedingErrorMessage = _.map( this.err && this.err.precedingErrors, function( error ) {
						return `\n\tPreceding error: ${ error.message }`;
					} ).join( "" );

					const message = util.format( "SqlContext Error. Failed on step \"%s\" with: \"%s\"%s", this.priorState, this.err.message, precedingErrorMessage );
					log( message );
					this.err.message = message;
					this.err.step = this.priorState;
					this.emit( "error", this.err );
				}
			}
		},

		step( alias, stepAction ) {
			let opt;
			if ( typeof stepAction === "object" ) {
				opt = stepAction;
				stepAction = function( execute ) {
					return execute( opt );
				};
			}
			addState( this, alias, stepAction );
			return this;
		},

		deferredStart() {
			if ( !this._started ) {
				process.nextTick( function() {
					this.handle( "start" );
				}.bind( this ) );
				this._started = true;
			}
		},

		end( fn ) {
			this.on( "end", fn );
			this.deferredStart();
			return this;
		},

	fsm.pipeline.push( name );
	fsm.states[ name ] = {
		_onEnter: function() {
			var exec = function( options ) {
				executeSql( fsm, name, options )
					.then(
						fsm.handle.bind( fsm, "success" ),
						fsm.handle.bind( fsm, "error" )
					);
			};
			stepAction.call(
				fsm,
				exec,
				fsm.results
			);
		},
		success: function( result ) {
			fsm.results[ name ] = result;

<MSG> Merge pull request #39 from digitalBush/feature-step-skip

Allow step to bypass execute callback.
<DFF> @@ -107,18 +107,36 @@ function addState( fsm, name, stepAction ) {
 	fsm.pipeline.push( name );
 	fsm.states[ name ] = {
 		_onEnter: function() {
+			var execCalled = false;
+			var skipped = false;
 			var exec = function( options ) {
+				if ( skipped ) {
+					var err = new Error( "Step '" + name + "' has already been skipped. Did you forget to return a promise?" );
+					errorHandler.call( fsm, err ); // Attempt to reject promise for sql transaction
+					throw err; // Reject the runaway promise
+				}
+
+				execCalled = true;
 				executeSql( fsm, name, options )
 					.then(
 						fsm.handle.bind( fsm, "success" ),
 						fsm.handle.bind( fsm, "error" )
 					);
 			};
-			stepAction.call(
+
+			var stepReturnValue = stepAction.call(
 				fsm,
 				exec,
 				fsm.results
 			);
+
+			when( stepReturnValue )
+				.then( function() {
+					if ( !execCalled ) {
+						skipped = true;
+						fsm.nextState();
+					}
+				}, fsm.handle.bind( fsm, "error" ) );
 		},
 		success: function( result ) {
 			fsm.results[ name ] = result;

{'commit': 'diff --git a/spec/behavior/sqlContext.spec.js b/spec/behavior/sqlContext.spec.js\nindex d2ea422..1a256d5 100644\n--- a/spec/behavior/sqlContext.spec.js\n+++ b/spec/behavior/sqlContext.spec.js\n@@ -1,5 +1,6 @@\n require( "../setup" );\n var mockConnectionFn = require( "../data/mockConnection" );\n+var when = require( "when" );\n \n /***************************************************\n \n@@ -479,6 +480,85 @@ describe( "SqlContext", function() {\n \t\t} );\n \t} );\n \n+\tdescribe( "when skipping execute call on a step", function() {\n+\t\tvar ctx;\n+\t\tbefore( function() {\n+\t\t\tsetup();\n+\n+\t\t\treqMock.expects( "query" )\n+\t\t\t\t.withArgs( "select * from sys.tables" )\n+\t\t\t\t.callsArgWith( 1, null, fakeRecords )\n+\t\t\t\t.once();\n+\n+\t\t\tctx = seriate.getPlainContext();\n+\t\t\treturn ctx\n+\t\t\t\t.step( "skipped", function( execute ) {\n+\t\t\t\t\treturn;\n+\t\t\t\t} )\n+\t\t\t\t.step( "read", {\n+\t\t\t\t\tquery: "select * from sys.tables"\n+\t\t\t\t} );\n+\t\t} );\n+\n+\t\tit( "should continue to next step", function() {\n+\t\t\treqMock.verify();\n+\t\t} );\n+\t} );\n+\n+\tdescribe( "when failing to return promise", function() {\n+\t\tvar runaway;\n+\t\tbefore( function() {\n+\t\t\tsetup();\n+\n+\t\t\treqMock.expects( "query" )\n+\t\t\t\t.withArgs( "select * from sys.tables" )\n+\t\t\t\t.never();\n+\n+\t\t\treturn seriate.getPlainContext()\n+\t\t\t\t.step( "badPromise", function( execute ) {\n+\t\t\t\t\trunaway = when( true )\n+\t\t\t\t\t\t.delay( 1 ) // Here to delay the call to execute but allow fsm to transition to success state\n+\t\t\t\t\t\t.then( function() {\n+\t\t\t\t\t\t\texecute( {\n+\t\t\t\t\t\t\t\tquery: "select * from sys.tables"\n+\t\t\t\t\t\t\t} );\n+\t\t\t\t\t\t} );\n+\t\t\t\t} );\n+\t\t} );\n+\n+\t\tit( "should throw error", function( done ) {\n+\t\t\treturn runaway\n+\t\t\t\t.catch( function( err ) {\n+\t\t\t\t\terr.should.be.instanceof( Error );\n+\t\t\t\t\tdone();\n+\t\t\t\t} );\n+\t\t} );\n+\n+\t\tit( "should not invoke sql", function( ) {\n+\t\t\treqMock.verify();\n+\t\t} );\n+\t} );\n+\n+\tdescribe( "when a promise rejects", function() {\n+\t\tvar error;\n+\t\tbefore( function() {\n+\t\t\tsetup();\n+\n+\t\t\treturn seriate.getPlainContext()\n+\t\t\t\t.step( "brokenPromise", function( execute ) {\n+\t\t\t\t\treturn when.reject( new Error( "NOPE!" ) );\n+\t\t\t\t} )\n+\t\t\t\t.then( null, function( err ) {\n+\t\t\t\t\terror = err;\n+\t\t\t\t} );\n+\t\t} );\n+\n+\t\tit( "should throw error", function( ) {\n+\t\t\terror.should.be.instanceof( Error );\n+\t\t\terror.message.should.contain( "NOPE!" );\n+\t\t} );\n+\t} );\n+\n \tdescribe( "with metrics", function() {\n \t\tdescribe( "when executing a query", function() {\n \t\t\tvar metrics, adapter;\ndiff --git a/src/sqlContext.js b/src/sqlContext.js\nindex 5a5a81c..3a69b57 100644\n--- a/src/sqlContext.js\n+++ b/src/sqlContext.js\n@@ -107,18 +107,36 @@ function addState( fsm, name, stepAction ) {\n \tfsm.pipeline.push( name );\n \tfsm.states[ name ] = {\n \t\t_onEnter: function() {\n+\t\t\tvar execCalled = false;\n+\t\t\tvar skipped = false;\n \t\t\tvar exec = function( options ) {\n+\t\t\t\tif ( skipped ) {\n+\t\t\t\t\tvar err = new Error( "Step \'" + name + "\' has already been skipped. Did you forget to return a promise?" );\n+\t\t\t\t\terrorHandler.call( fsm, err ); // Attempt to reject promise for sql transaction\n+\t\t\t\t\tthrow err; // Reject the runaway promise\n+\t\t\t\t}\n+\n+\t\t\t\texecCalled = true;\n \t\t\t\texecuteSql( fsm, name, options )\n \t\t\t\t\t.then(\n \t\t\t\t\t\tfsm.handle.bind( fsm, "success" ),\n \t\t\t\t\t\tfsm.handle.bind( fsm, "error" )\n \t\t\t\t\t);\n \t\t\t};\n-\t\t\tstepAction.call(\n+\n+\t\t\tvar stepReturnValue = stepAction.call(\n \t\t\t\tfsm,\n \t\t\t\texec,\n \t\t\t\tfsm.results\n \t\t\t);\n+\n+\t\t\twhen( stepReturnValue )\n+\t\t\t\t.then( function() {\n+\t\t\t\t\tif ( !execCalled ) {\n+\t\t\t\t\t\tskipped = true;\n+\t\t\t\t\t\tfsm.nextState();\n+\t\t\t\t\t}\n+\t\t\t\t}, fsm.handle.bind( fsm, "error" ) );\n \t\t},\n \t\tsuccess: function( result ) {\n \t\t\tfsm.results[ name ] = result;\n', 'commit_subject': 'Merge pull request #39 from digitalBush/feature-step-skip', 'repo_name': 'LeanKit-Labs/seriate', 'committer': {'name': 'Doug Neiner', 'email': '0798ebecfb7bc4dfb9d0445c2fa030eb64e1fa1c@dougneiner.com', 'time_sec': '1458598721', 'tz_offset': '-300', 'date': {'seconds': '1458598721'}}, 'commit_tree': 'ea5b209144b2e0afa3a72a476200cd5a5402da5d', 'commit_message': 'Merge pull request #39 from digitalBush/feature-step-skip\n\nAllow step to bypass execute callback.', 'language_name': 'JavaScript', 'license': 'mit', 'hunks': '@@ -107,18 +107,36 @@ function addState( fsm, name, stepAction ) {\n \tfsm.pipeline.push( name );\n \tfsm.states[ name ] = {\n \t\t_onEnter: function() {\n+\t\t\tvar execCalled = false;\n+\t\t\tvar skipped = false;\n \t\t\tvar exec = function( options ) {\n+\t\t\t\tif ( skipped ) {\n+\t\t\t\t\tvar err = new Error( "Step \'" + name + "\' has already been skipped. Did you forget to return a promise?" );\n+\t\t\t\t\terrorHandler.call( fsm, err ); // Attempt to reject promise for sql transaction\n+\t\t\t\t\tthrow err; // Reject the runaway promise\n+\t\t\t\t}\n+\n+\t\t\t\texecCalled = true;\n \t\t\t\texecuteSql( fsm, name, options )\n \t\t\t\t\t.then(\n \t\t\t\t\t\tfsm.handle.bind( fsm, "success" ),\n \t\t\t\t\t\tfsm.handle.bind( fsm, "error" )\n \t\t\t\t\t);\n \t\t\t};\n-\t\t\tstepAction.call(\n+\n+\t\t\tvar stepReturnValue = stepAction.call(\n \t\t\t\tfsm,\n \t\t\t\texec,\n \t\t\t\tfsm.results\n \t\t\t);\n+\n+\t\t\twhen( stepReturnValue )\n+\t\t\t\t.then( function() {\n+\t\t\t\t\tif ( !execCalled ) {\n+\t\t\t\t\t\tskipped = true;\n+\t\t\t\t\t\tfsm.nextState();\n+\t\t\t\t\t}\n+\t\t\t\t}, fsm.handle.bind( fsm, "error" ) );\n \t\t},\n \t\tsuccess: function( result ) {\n \t\t\tfsm.results[ name ] = result;\n', 'addition_count': 19, 'deletion_count': 1, 'src_file': 'a/src/sqlContext.js', 'tgt_file': 'b/src/sqlContext.js', 'file_extension': '.js', 'before_file': 'const _ = require( "lodash" );\nconst util = require( "util" );\nconst EventEmitter = require( "events" );\n\nconst machina = require( "machina" );\nconst log = require( "debug" )( "seriate:sql" );\nconst addState = require( "./sqlContextUtils" ).addState;\n\nmodule.exports = function() {\n\tconst SqlContext = machina.Fsm.extend( {\n\t\t_connected( connection ) {\n\t\t\tthis.connection = connection;\n\t\t\tthis.handle( "success" );\n\t\t},\n\n\t\t_connectionError( err ) {\n\t\t\tthis.handle( "error", err );\n\t\t},\n\n\t\tinitialState: "uninitialized",\n\n\t\tinitialize( options ) {\n\t\t\tthis.results = {};\n\t\t\tthis.pipeline = [];\n\t\t\tthis.pipePos = -1;\n\t\t\tthis.metrics = options.metrics;\n\t\t\tthis.metricsNamespace = options.namespace;\n\t\t\tthis.options = options;\n\t\t\toptions.connection\n\t\t\t\t.then(\n\t\t\t\t\tthis._connected.bind( this ),\n\t\t\t\t\tthis._connectionError.bind( this )\n\t\t\t\t);\n\t\t},\n\n\t\tnextState() {\n\t\t\tthis.pipePos += 1;\n\t\t\tconst nextState = this.pipeline[ this.pipePos ] || "done";\n\t\t\tthis.transition( nextState );\n\t\t},\n\n\t\tstates: {\n\t\t\tuninitialized: {\n\t\t\t\tstart: "connecting",\n\t\t\t\t"*"() {\n\t\t\t\t\tthis.deferUntilTransition( "connecting" );\n\t\t\t\t}\n\t\t\t},\n\t\t\tconnecting: {\n\t\t\t\tsuccess() {\n\t\t\t\t\tthis.nextState();\n\t\t\t\t},\n\t\t\t\terror( err ) {\n\t\t\t\t\tthis.err = err;\n\t\t\t\t\tthis.transition( "error" );\n\t\t\t\t}\n\t\t\t},\n\n\t\t\tdone: {\n\t\t\t\t_onEnter() {\n\t\t\t\t\tthis.emit( "end", this.results );\n\t\t\t\t}\n\t\t\t},\n\n\t\t\terror: {\n\t\t\t\t_onEnter() {\n\t\t\t\t\tconst precedingErrorMessage = _.map( this.err && this.err.precedingErrors, function( error ) {\n\t\t\t\t\t\treturn `\\n\\tPreceding error: ${ error.message }`;\n\t\t\t\t\t} ).join( "" );\n\n\t\t\t\t\tconst message = util.format( "SqlContext Error. Failed on step \\"%s\\" with: \\"%s\\"%s", this.priorState, this.err.message, precedingErrorMessage );\n\t\t\t\t\tlog( message );\n\t\t\t\t\tthis.err.message = message;\n\t\t\t\t\tthis.err.step = this.priorState;\n\t\t\t\t\tthis.emit( "error", this.err );\n\t\t\t\t}\n\t\t\t}\n\t\t},\n\n\t\tstep( alias, stepAction ) {\n\t\t\tlet opt;\n\t\t\tif ( typeof stepAction === "object" ) {\n\t\t\t\topt = stepAction;\n\t\t\t\tstepAction = function( execute ) {\n\t\t\t\t\treturn execute( opt );\n\t\t\t\t};\n\t\t\t}\n\t\t\taddState( this, alias, stepAction );\n\t\t\treturn this;\n\t\t},\n\n\t\tdeferredStart() {\n\t\t\tif ( !this._started ) {\n\t\t\t\tprocess.nextTick( function() {\n\t\t\t\t\tthis.handle( "start" );\n\t\t\t\t}.bind( this ) );\n\t\t\t\tthis._started = true;\n\t\t\t}\n\t\t},\n\n\t\tend( fn ) {\n\t\t\tthis.on( "end", fn );\n\t\t\tthis.deferredStart();\n\t\t\treturn this;\n\t\t},\n\n\tfsm.pipeline.push( name );\n\tfsm.states[ name ] = {\n\t\t_onEnter: function() {\n\t\t\tvar exec = function( options ) {\n\t\t\t\texecuteSql( fsm, name, options )\n\t\t\t\t\t.then(\n\t\t\t\t\t\tfsm.handle.bind( fsm, "success" ),\n\t\t\t\t\t\tfsm.handle.bind( fsm, "error" )\n\t\t\t\t\t);\n\t\t\t};\n\t\t\tstepAction.call(\n\t\t\t\tfsm,\n\t\t\t\texec,\n\t\t\t\tfsm.results\n\t\t\t);\n\t\t},\n\t\tsuccess: function( result ) {\n\t\t\tfsm.results[ name ] = result;\n', 'message': 'Merge pull request #39 from digitalBush/feature-step-skip\n\nAllow step to bypass execute callback.', 'author_name': 'Doug Neiner', 'lang': 'js'}