Piping baby

pipe.js

	/*
	 * Sets up control flows involving multiple callbacks and dependencies
	 * Usage:
	 * var p = Q.pipe(function (params, subjects) {
	 *   // arguments that were passed are in params.user, params.stream
	 *   // this objects that were passed are in subjects.user, subjects.stream
	 * }, ['user', 'stream]);
	 * mysql("SELECT * FROM user WHERE user_id = 2", p.fill('user'));
	 * mysql("SELECT * FROM stream WHERE publisher_id = 2", p.fill('stream'));
	 *
	 * @param callback Function
	 *  Once all required fields are filled (see the next parameter)
	 *  this function is called every time something is piped.
	 *  If you return false from this function, it will no longer be called
	 *  for future pipes.
	 * @param requires Array
	 *  Optional. Pass an array of required field names here.
	 * @paraam callback2 Function
	 * @param requires2 Array
	 *  Keep passing as many functions (followed by arrays of required fields)
	 *  as you want and they will be processed in order every time something is
	 *  piped in. You would typically do this to set up some functions
	 *  that depend on some fields, and other functions that depend on other fields
	 *  and have functions execute once all the data is available.
	 *  Thus you can mix and match sequential and parallel processing.
	 *  If you need getters, use Q.getter( ).
	 *  If you need to do throttling, use Q.Throttle.
	 * @return Object
	 *  An object with the following method: fill(field).
	 *  Call this method to return a callback.
	 */
	me.pipe = function(callback, requires) {
		var cb, callbacks=[];
		for (var i=0; i<arguments.length; ++i) {
			if (arguments[i] instanceof Function) {
				callbacks.push(cb = arguments[i]);
			} else if (cb && arguments[i] instanceof Array) {
				cb.requires = arguments[i];
			}
		}
		var result = {
			callbacks: callbacks,
			params: {},
			subjects: {},
			fill: function(field, defaultReturn) {
				var t = this;
				return function() {
					t.params[field] = Array.prototype.slice.call(arguments);
					t.subjects[field] = this;
					t.handle();
					return defaultReturn;
				};
			},
			handle: function () {
				var cb, found;
				cbloop:
				for (var i=0; i<callbacks.length; ++i) {
					cb = callbacks[i];
					if (cb.stopCalling) {
						continue;
					}
					if (cb.requires) {
						for (var j=0; j<cb.requires.length; ++j) {
							if (! (cb.requires[j] in this.params)) {
								continue cbloop;
							}
						}
					}
					if (cb.call(this, this.params, this.subjects) === false) {
						cb.stopCalling = true;
					}
				}
			}
		};
		return result;
	};