|
|
- var streamify = require('./streamify.js')
- , defer = require('./defer.js')
- ;
-
- // API
- module.exports = ReadableAsyncKit;
-
- /**
- * Base constructor for all streams
- * used to hold properties/methods
- */
- function ReadableAsyncKit()
- {
- ReadableAsyncKit.super_.apply(this, arguments);
-
- // list of active jobs
- this.jobs = {};
-
- // add stream methods
- this.destroy = destroy;
- this._start = _start;
- this._read = _read;
- }
-
- /**
- * Destroys readable stream,
- * by aborting outstanding jobs
- *
- * @returns {void}
- */
- function destroy()
- {
- if (this.destroyed)
- {
- return;
- }
-
- this.destroyed = true;
-
- if (typeof this.terminator == 'function')
- {
- this.terminator();
- }
- }
-
- /**
- * Starts provided jobs in async manner
- *
- * @private
- */
- function _start()
- {
- // first argument – runner function
- var runner = arguments[0]
- // take away first argument
- , args = Array.prototype.slice.call(arguments, 1)
- // second argument - input data
- , input = args[0]
- // last argument - result callback
- , endCb = streamify.callback.call(this, args[args.length - 1])
- ;
-
- args[args.length - 1] = endCb;
- // third argument - iterator
- args[1] = streamify.iterator.call(this, args[1]);
-
- // allow time for proper setup
- defer(function()
- {
- if (!this.destroyed)
- {
- this.terminator = runner.apply(null, args);
- }
- else
- {
- endCb(null, Array.isArray(input) ? [] : {});
- }
- }.bind(this));
- }
-
-
- /**
- * Implement _read to comply with Readable streams
- * Doesn't really make sense for flowing object mode
- *
- * @private
- */
- function _read()
- {
-
- }
|