Update to NPM version

ProjectSourceCode/node_modules/spex/lib/ext/batch.js

@@ -0,0 +1,151 @@
+const {BatchError} = require('../errors/batch');
+
+/**
+ * @method batch
+ * @description
+ * Settles (resolves or rejects) every [mixed value]{@tutorial mixed} in the input array.
+ *
+ * The method resolves with an array of results, the same as the standard $[promise.all],
+ * while providing comprehensive error details in case of a reject, in the form of
+ * type {@link errors.BatchError BatchError}.
+ *
+ * @param {Array} values
+ * Array of [mixed values]{@tutorial mixed} (it can be empty), to be resolved asynchronously, in no particular order.
+ *
+ * Passing in anything other than an array will reject with {@link external:TypeError TypeError} =
+ * `Method 'batch' requires an array of values.`
+ *
+ * @param {Object} [options]
+ * Optional Parameters.
+ *
+ * @param {Function|generator} [options.cb]
+ * Optional callback (or generator) to receive the result for each settled value.
+ *
+ * Callback Parameters:
+ *  - `index` = index of the value in the source array
+ *  - `success` - indicates whether the value was resolved (`true`), or rejected (`false`)
+ *  - `result` = resolved data, if `success`=`true`, or else the rejection reason
+ *  - `delay` = number of milliseconds since the last call (`undefined` when `index=0`)
+ *
+ * The function inherits `this` context from the calling method.
+ *
+ * It can optionally return a promise to indicate that notifications are handled asynchronously.
+ * And if the returned promise resolves, it signals a successful handling, while any resolved
+ * data is ignored.
+ *
+ * If the function returns a rejected promise or throws an error, the entire method rejects
+ * with {@link errors.BatchError BatchError} where the corresponding value in property `data`
+ * is set to `{success, result, origin}`:
+ *  - `success` = `false`
+ *  - `result` = the rejection reason or the error thrown by the notification callback
+ *  - `origin` = the original data passed into the callback as object `{success, result}`
+ *
+ * @returns {external:Promise}
+ *
+ * The method resolves with an array of individual resolved results, the same as the standard $[promise.all].
+ * In addition, the array is extended with a hidden read-only property `duration` - number of milliseconds
+ * spent resolving all the data.
+ *
+ * The method rejects with {@link errors.BatchError BatchError} when any of the following occurs:
+ *  - one or more values rejected or threw an error while being resolved as a [mixed value]{@tutorial mixed}
+ *  - notification callback `cb` returned a rejected promise or threw an error
+ *
+ */
+function batch(values, options, config) {
+
+    const $p = config.promise, utils = config.utils;
+
+    if (!Array.isArray(values)) {
+        return $p.reject(new TypeError('Method \'batch\' requires an array of values.'));
+    }
+
+    if (!values.length) {
+        const empty = [];
+        utils.extend(empty, 'duration', 0);
+        return $p.resolve(empty);
+    }
+
+    options = options || {};
+
+    const cb = utils.wrap(options.cb),
+        self = this, start = Date.now();
+
+    return $p((resolve, reject) => {
+        let cbTime, remaining = values.length;
+        const errors = [], result = new Array(remaining);
+        values.forEach((item, i) => {
+            utils.resolve.call(self, item, null, data => {
+                result[i] = data;
+                step(i, true, data);
+            }, reason => {
+                result[i] = {success: false, result: reason};
+                errors.push(i);
+                step(i, false, reason);
+            });
+        });
+
+        function step(idx, pass, data) {
+            if (cb) {
+                const cbNow = Date.now(),
+                    cbDelay = idx ? (cbNow - cbTime) : undefined;
+                let cbResult;
+                cbTime = cbNow;
+                try {
+                    cbResult = cb.call(self, idx, pass, data, cbDelay);
+                } catch (e) {
+                    setError(e);
+                }
+                if (utils.isPromise(cbResult)) {
+                    cbResult
+                        .then(check)
+                        .catch(error => {
+                            setError(error);
+                            check();
+                        });
+                } else {
+                    check();
+                }
+            } else {
+                check();
+            }
+
+            function setError(e) {
+                const r = pass ? {success: false} : result[idx];
+                if (pass) {
+                    result[idx] = r;
+                    errors.push(idx);
+                }
+                r.result = e;
+                r.origin = {success: pass, result: data};
+            }
+
+            function check() {
+                if (!--remaining) {
+                    if (errors.length) {
+                        errors.sort();
+                        if (errors.length < result.length) {
+                            for (let i = 0, k = 0; i < result.length; i++) {
+                                if (i === errors[k]) {
+                                    k++;
+                                } else {
+                                    result[i] = {success: true, result: result[i]};
+                                }
+                            }
+                        }
+                        reject(new BatchError(result, errors, Date.now() - start));
+                    } else {
+                        utils.extend(result, 'duration', Date.now() - start);
+                        resolve(result);
+                    }
+                }
+                return null; // this dummy return is just to prevent Bluebird warnings;
+            }
+        }
+    });
+}
+
+module.exports = function (config) {
+    return function (values, options) {
+        return batch.call(this, values, options, config);
+    };
+};

ProjectSourceCode/node_modules/spex/lib/ext/page.js

@@ -0,0 +1,182 @@
+const {PageError} = require('../errors/page');
+
+/**
+ * @method page
+ * @description
+ * Resolves a dynamic sequence of pages/arrays with [mixed values]{@tutorial mixed}.
+ *
+ * The method acquires pages (arrays of [mixed values]{@tutorial mixed}) from the `source` function, one by one,
+ * and resolves each page as a {@link batch}, till no more pages left or an error/reject occurs.
+ *
+ * @param {Function|generator} source
+ * Expected to return a [mixed value]{@tutorial mixed} that resolves with the next page of data (array of [mixed values]{@tutorial mixed}).
+ * Returning or resolving with `undefined` ends the sequence, and the method resolves.
+ *
+ * The function inherits `this` context from the calling method.
+ *
+ * Parameters:
+ *  - `index` = index of the page being requested
+ *  - `data` = previously returned page, resolved as a {@link batch} (`undefined` when `index=0`)
+ *  - `delay` = number of milliseconds since the last call (`undefined` when `index=0`)
+ *
+ * If the function throws an error or returns a rejected promise, the method rejects with
+ * {@link errors.PageError PageError}, which will have property `source` set.
+ *
+ * And if the function returns or resolves with anything other than an array or `undefined`,
+ * the method rejects with the same {@link errors.PageError PageError}, but with `error` set to
+ * `Unexpected data returned from the source.`
+ *
+ * Passing in anything other than a function will reject with {@link external:TypeError TypeError} = `Parameter 'source' must be a function.`
+ *
+ * @param {Object} [options]
+ * Optional Parameters.
+ *
+ * @param {Function|generator} [options.dest]
+ * Optional destination function (or generator), to receive a resolved {@link batch} of data
+ * for each page, process it and respond as required.
+ *
+ * Parameters:
+ *  - `index` = page index in the sequence
+ *  - `data` = page data resolved as a {@link batch}
+ *  - `delay` = number of milliseconds since the last call (`undefined` when `index=0`)
+ *
+ * The function inherits `this` context from the calling method.
+ *
+ * It can optionally return a promise object, if notifications are handled asynchronously.
+ * And if a promise is returned, the method will not request another page from the `source`
+ * function until the promise has been resolved.
+ *
+ * If the function throws an error or returns a rejected promise, the sequence terminates,
+ * and the method rejects with {@link errors.PageError PageError}, which will have property `dest` set.
+ *
+ * @param {Number} [options.limit=0]
+ * Limits the maximum number of pages to be requested from the `source`. If the value is greater
+ * than 0, the method will successfully resolve once the specified limit has been reached.
+ *
+ * When `limit` isn't specified (default), the sequence is unlimited, and it will continue
+ * till one of the following occurs:
+ *  - `source` returns or resolves with `undefined` or an invalid value (non-array)
+ *  - either `source` or `dest` functions throw an error or return a rejected promise
+ *
+ * @returns {external:Promise}
+ *
+ * When successful, the method resolves with object `{pages, total, duration}`:
+ *  - `pages` = number of pages resolved
+ *  - `total` = the sum of all page sizes (total number of values resolved)
+ *  - `duration` = number of milliseconds consumed by the method
+ *
+ * When the method fails, it rejects with {@link errors.PageError PageError}.
+ *
+ */
+function page(source, options, config) {
+
+    const $p = config.promise, spex = config.spex, utils = config.utils;
+
+    if (typeof source !== 'function') {
+        return $p.reject(new TypeError('Parameter \'source\' must be a function.'));
+    }
+
+    options = options || {};
+    source = utils.wrap(source);
+
+    const limit = (options.limit > 0) ? parseInt(options.limit) : 0,
+        dest = utils.wrap(options.dest), self = this, start = Date.now();
+    let request, srcTime, destTime, total = 0;
+
+    return $p((resolve, reject) => {
+
+        function loop(idx) {
+            const srcNow = Date.now(),
+                srcDelay = idx ? (srcNow - srcTime) : undefined;
+            srcTime = srcNow;
+            utils.resolve.call(self, source, [idx, request, srcDelay], value => {
+                if (value === undefined) {
+                    success();
+                } else {
+                    if (value instanceof Array) {
+                        spex.batch(value)
+                            .then(data => {
+                                request = data;
+                                total += data.length;
+                                if (dest) {
+                                    const destNow = Date.now(),
+                                        destDelay = idx ? (destNow - destTime) : undefined;
+                                    let destResult;
+                                    destTime = destNow;
+                                    try {
+                                        destResult = dest.call(self, idx, data, destDelay);
+                                    } catch (err) {
+                                        fail({
+                                            error: err,
+                                            dest: data
+                                        }, 4, dest.name);
+                                        return;
+                                    }
+                                    if (utils.isPromise(destResult)) {
+                                        destResult
+                                            .then(next)
+                                            .catch(error => {
+                                                fail({
+                                                    error: error,
+                                                    dest: data
+                                                }, 3, dest.name);
+                                            });
+                                    } else {
+                                        next();
+                                    }
+                                } else {
+                                    next();
+                                }
+                                return null; // this dummy return is just to prevent Bluebird warnings;
+                            })
+                            .catch(error => {
+                                fail({
+                                    error: error
+                                }, 0);
+                            });
+                    } else {
+                        fail({
+                            error: new Error('Unexpected data returned from the source.'),
+                            source: request
+                        }, 5, source.name);
+                    }
+                }
+            }, (reason, isRej) => {
+                fail({
+                    error: reason,
+                    source: request
+                }, isRej ? 1 : 2, source.name);
+            });
+
+            function next() {
+                if (limit === ++idx) {
+                    success();
+                } else {
+                    loop(idx);
+                }
+                return null; // this dummy return is just to prevent Bluebird warnings;
+            }
+
+            function success() {
+                resolve({
+                    pages: idx,
+                    total: total,
+                    duration: Date.now() - start
+                });
+            }
+
+            function fail(reason, code, cbName) {
+                reason.index = idx;
+                reject(new PageError(reason, code, cbName, Date.now() - start));
+            }
+        }
+
+        loop(0);
+    });
+}
+
+module.exports = function (config) {
+    return function (source, options) {
+        return page.call(this, source, options, config);
+    };
+};

ProjectSourceCode/node_modules/spex/lib/ext/sequence.js

@@ -0,0 +1,194 @@
+const {SequenceError} = require('../errors/sequence');
+
+/**
+ * @method sequence
+ * @description
+ * Resolves a dynamic sequence of [mixed values]{@tutorial mixed}.
+ *
+ * The method acquires [mixed values]{@tutorial mixed} from the `source` function, one at a time, and resolves them,
+ * till either no more values left in the sequence or an error/reject occurs.
+ *
+ * It supports both [linked and detached sequencing]{@tutorial sequencing}.
+ *
+ * @param {Function|generator} source
+ * Expected to return the next [mixed value]{@tutorial mixed} to be resolved. Returning or resolving
+ * with `undefined` ends the sequence, and the method resolves.
+ *
+ * Parameters:
+ *  - `index` = current request index in the sequence
+ *  - `data` = resolved data from the previous call (`undefined` when `index=0`)
+ *  - `delay` = number of milliseconds since the last call (`undefined` when `index=0`)
+ *
+ * The function inherits `this` context from the calling method.
+ *
+ * If the function throws an error or returns a rejected promise, the sequence terminates,
+ * and the method rejects with {@link errors.SequenceError SequenceError}, which will have property `source` set.
+ *
+ * Passing in anything other than a function will reject with {@link external:TypeError TypeError} = `Parameter 'source' must be a function.`
+ *
+ * @param {Object} [options]
+ * Optional Parameters.
+ *
+ * @param {Function|generator} [options.dest=null]
+ * Optional destination function (or generator), to receive resolved data for each index,
+ * process it and respond as required.
+ *
+ * Parameters:
+ *  - `index` = index of the resolved data in the sequence
+ *  - `data` = the data resolved
+ *  - `delay` = number of milliseconds since the last call (`undefined` when `index=0`)
+ *
+ * The function inherits `this` context from the calling method.
+ *
+ * It can optionally return a promise object, if data processing is done asynchronously.
+ * If a promise is returned, the method will not request another value from the `source` function,
+ * until the promise has been resolved (the resolved value is ignored).
+ *
+ * If the function throws an error or returns a rejected promise, the sequence terminates,
+ * and the method rejects with {@link errors.SequenceError SequenceError}, which will have property `dest` set.
+ *
+ * @param {Number} [options.limit=0]
+ * Limits the maximum size of the sequence. If the value is greater than 0, the method will
+ * successfully resolve once the specified limit has been reached.
+ *
+ * When `limit` isn't specified (default), the sequence is unlimited, and it will continue
+ * till one of the following occurs:
+ *  - `source` either returns or resolves with `undefined`
+ *  - either `source` or `dest` functions throw an error or return a rejected promise
+ *
+ * @param {Boolean} [options.track=false]
+ * Changes the type of data to be resolved by this method. By default, it is `false`
+ * (see the return result). When set to be `true`, the method tracks/collects all resolved data
+ * into an array internally, and resolves with that array once the method has finished successfully.
+ *
+ * It must be used with caution, as to the size of the sequence, because accumulating data for
+ * a very large sequence can result in consuming too much memory.
+ *
+ * @returns {external:Promise}
+ *
+ * When successful, the resolved data depends on parameter `track`. When `track` is `false`
+ * (default), the method resolves with object `{total, duration}`:
+ *  - `total` = number of values resolved by the sequence
+ *  - `duration` = number of milliseconds consumed by the method
+ *
+ * When `track` is `true`, the method resolves with an array of all the data that has been resolved,
+ * the same way that the standard $[promise.all] resolves. In addition, the array comes extended with
+ * a hidden read-only property `duration` - number of milliseconds consumed by the method.
+ *
+ * When the method fails, it rejects with {@link errors.SequenceError SequenceError}.
+ */
+function sequence(source, options, config) {
+
+    const $p = config.promise, utils = config.utils;
+
+    if (typeof source !== 'function') {
+        return $p.reject(new TypeError('Parameter \'source\' must be a function.'));
+    }
+
+    source = utils.wrap(source);
+
+    options = options || {};
+
+    const limit = (options.limit > 0) ? parseInt(options.limit) : 0,
+        dest = utils.wrap(options.dest),
+        self = this, start = Date.now();
+    let data, srcTime, destTime, result = [];
+
+    return $p((resolve, reject) => {
+
+        function loop(idx) {
+            const srcNow = Date.now(),
+                srcDelay = idx ? (srcNow - srcTime) : undefined;
+            srcTime = srcNow;
+            utils.resolve.call(self, source, [idx, data, srcDelay], (value, delayed) => {
+                data = value;
+                if (data === undefined) {
+                    success();
+                } else {
+                    if (options.track) {
+                        result.push(data);
+                    }
+                    if (dest) {
+                        const destNow = Date.now(),
+                            destDelay = idx ? (destNow - destTime) : undefined;
+                        let destResult;
+                        destTime = destNow;
+                        try {
+                            destResult = dest.call(self, idx, data, destDelay);
+                        } catch (e) {
+                            fail({
+                                error: e,
+                                dest: data
+                            }, 3, dest.name);
+                            return;
+                        }
+                        if (utils.isPromise(destResult)) {
+                            destResult
+                                .then(() => {
+                                    next(true);
+                                    return null; // this dummy return is just to prevent Bluebird warnings;
+                                })
+                                .catch(error => {
+                                    fail({
+                                        error: error,
+                                        dest: data
+                                    }, 2, dest.name);
+                                });
+                        } else {
+                            next(delayed);
+                        }
+                    } else {
+                        next(delayed);
+                    }
+                }
+            }, (reason, isRej) => {
+                fail({
+                    error: reason,
+                    source: data
+                }, isRej ? 0 : 1, source.name);
+            });
+
+            function next(delayed) {
+                if (limit === ++idx) {
+                    success();
+                } else {
+                    if (delayed) {
+                        loop(idx);
+                    } else {
+                        $p.resolve()
+                            .then(() => {
+                                loop(idx);
+                                return null; // this dummy return is just to prevent Bluebird warnings;
+                            });
+                    }
+                }
+            }
+
+            function success() {
+                const length = Date.now() - start;
+                if (options.track) {
+                    utils.extend(result, 'duration', length);
+                } else {
+                    result = {
+                        total: idx,
+                        duration: length
+                    };
+                }
+                resolve(result);
+            }
+
+            function fail(reason, code, cbName) {
+                reason.index = idx;
+                reject(new SequenceError(reason, code, cbName, Date.now() - start));
+            }
+        }
+
+        loop(0);
+    });
+}
+
+module.exports = function (config) {
+    return function (source, options) {
+        return sequence.call(this, source, options, config);
+    };
+};

ProjectSourceCode/node_modules/spex/lib/ext/stream/index.js

@@ -0,0 +1,65 @@
+const npm = {
+    read: require('./read')
+};
+
+/**
+ * @namespace stream
+ * @description
+ * Namespace with methods that implement stream operations, and {@link stream.read read} is the only method currently supported.
+ *
+ * **Synchronous Stream Processing**
+ *
+ * ```js
+ * const stream = require('spex')(Promise).stream;
+ * const fs = require('fs');
+ *
+ * const rs = fs.createReadStream('values.txt');
+ *
+ * function receiver(index, data, delay) {
+ *    console.log('RECEIVED:', index, data, delay);
+ * }
+ *
+ * stream.read(rs, receiver)
+ *     .then(data => {
+ *         console.log('DATA:', data);
+ *     })
+ *     .catch(error => {
+ *         console.log('ERROR:', error);
+ *     });
+ * ```
+ *
+ * **Asynchronous Stream Processing**
+ *
+ * ```js
+ * const stream = require('spex')(Promise).stream;
+ * const fs = require('fs');
+ *
+ * const rs = fs.createReadStream('values.txt');
+ *
+ * function receiver(index, data, delay) {
+ *    return new Promise(resolve => {
+ *        console.log('RECEIVED:', index, data, delay);
+ *        resolve();
+ *    });
+ * }
+ *
+ * stream.read(rs, receiver)
+ *     .then(data => {
+ *         console.log('DATA:', data);
+ *     })
+ *     .catch(error => {
+ *         console.log('ERROR:', error);
+ *    });
+ * ```
+ *
+ * @property {function} stream.read
+ * Consumes and processes data from a $[Readable] stream.
+ *
+ */
+module.exports = function (config) {
+    const res = {
+        read: npm.read(config)
+    };
+    Object.freeze(res);
+    return res;
+};

ProjectSourceCode/node_modules/spex/lib/ext/stream/read.js

@@ -0,0 +1,206 @@
+/**
+ * @method stream.read
+ * @description
+ * Consumes and processes data from a $[Readable] stream.
+ *
+ * It reads the entire stream, using either **paused mode** (default), or in chunks (see `options.readChunks`)
+ * with support for both synchronous and asynchronous data processing.
+ *
+ * **NOTE:** Once the method has finished, the onus is on the caller to release the stream
+ * according to its protocol.
+ *
+ * @param {Object} stream
+ * $[Readable] stream object.
+ *
+ * Passing in anything else will throw `Readable stream is required.`
+ *
+ * @param {Function|generator} receiver
+ * Data processing callback (or generator).
+ *
+ * Passing in anything else will throw `Invalid stream receiver.`
+ *
+ * Parameters:
+ *  - `index` = index of the call made to the function
+ *  - `data` = array of all data reads from the stream's buffer
+ *  - `delay` = number of milliseconds since the last call (`undefined` when `index=0`)
+ *
+ * The function is called with the same `this` context as the calling method.
+ *
+ * It can optionally return a promise object, if data processing is asynchronous.
+ * And if a promise is returned, the method will not read data from the stream again,
+ * until the promise has been resolved.
+ *
+ * If the function throws an error or returns a rejected promise, the method rejects
+ * with the same error / rejection reason.
+ *
+ * @param {Object} [options]
+ * Optional Parameters.
+ *
+ * @param {Boolean} [options.closable=false]
+ * Instructs the method to resolve on event `close` supported by the stream, as opposed to event
+ * `end` that's used by default.
+ *
+ * @param {Boolean} [options.readChunks=false]
+ * By default, the method handles event `readable` of the stream to consume data in a simplified form,
+ * item by item. If you enable this option, the method will instead handle event `data` of the stream,
+ * to consume chunks of data.
+ *
+ * @param {Number} [options.readSize]
+ * When the value is greater than 0, it sets the read size from the stream's buffer
+ * when the next data is available. By default, the method uses as few reads as possible
+ * to get all the data currently available in the buffer.
+ *
+ * NOTE: This option is ignored when option `readChunks` is enabled.
+ *
+ * @returns {external:Promise}
+ *
+ * When finished successfully, resolves with object `{calls, reads, length, duration}`:
+ *  - `calls` = number of calls made into the `receiver`
+ *  - `reads` = number of successful reads from the stream
+ *  - `length` = total length for all the data reads from the stream
+ *  - `duration` = number of milliseconds consumed by the method
+ *
+ * When it fails, the method rejects with the error/reject specified,
+ * which can happen as a result of:
+ *  - event `error` emitted by the stream
+ *  - receiver throws an error or returns a rejected promise
+ */
+function read(stream, receiver, options, config) {
+
+    const $p = config.promise, utils = config.utils;
+
+    if (!utils.isReadableStream(stream)) {
+        return $p.reject(new TypeError('Readable stream is required.'));
+    }
+
+    if (typeof receiver !== 'function') {
+        return $p.reject(new TypeError('Invalid stream receiver.'));
+    }
+
+    receiver = utils.wrap(receiver);
+
+    options = options || {};
+
+    const readSize = (options.readSize > 0) ? parseInt(options.readSize) : null,
+        self = this, start = Date.now(), receiveEvent = options.readChunks ? 'data' : 'readable';
+    let cbTime, ready, waiting, stop, reads = 0, length = 0, index = 0;
+
+    return $p((resolve, reject) => {
+
+        function onReceive(data) {
+            ready = true;
+            process(data);
+        }
+
+        function onEnd() {
+            if (!options.closable) {
+                success();
+            }
+        }
+
+        function onClose() {
+            success();
+        }
+
+        function onError(error) {
+            fail(error);
+        }
+
+        stream.on(receiveEvent, onReceive);
+        stream.on('end', onEnd);
+        stream.on('close', onClose);
+        stream.on('error', onError);
+
+        function process(data) {
+            if (!ready || stop || waiting) {
+                return;
+            }
+            ready = false;
+            let cache;
+            if (options.readChunks) {
+                cache = data;
+                // istanbul ignore else;
+                // we cannot test the else condition, as it requires a special broken stream interface.
+                if (!Array.isArray(cache)) {
+                    cache = [cache];
+                }
+                length += cache.length;
+                reads++;
+            } else {
+                cache = [];
+                waiting = true;
+                let page;
+                do {
+                    page = stream.read(readSize);
+                    if (page) {
+                        cache.push(page);
+                        // istanbul ignore next: requires a unique stream that
+                        // creates objects without property `length` defined.
+                        length += page.length || 0;
+                        reads++;
+                    }
+                } while (page);
+
+                if (!cache.length) {
+                    waiting = false;
+                    return;
+                }
+            }
+
+            const cbNow = Date.now(),
+                cbDelay = index ? (cbNow - cbTime) : undefined;
+            let result;
+            cbTime = cbNow;
+            try {
+                result = receiver.call(self, index++, cache, cbDelay);
+            } catch (e) {
+                fail(e);
+                return;
+            }
+
+            if (utils.isPromise(result)) {
+                result
+                    .then(() => {
+                        waiting = false;
+                        process();
+                        return null; // this dummy return is just to prevent Bluebird warnings;
+                    })
+                    .catch(error => {
+                        fail(error);
+                    });
+            } else {
+                waiting = false;
+                process();
+            }
+        }
+
+        function success() {
+            cleanup();
+            resolve({
+                calls: index,
+                reads: reads,
+                length: length,
+                duration: Date.now() - start
+            });
+        }
+
+        function fail(error) {
+            stop = true;
+            cleanup();
+            reject(error);
+        }
+
+        function cleanup() {
+            stream.removeListener(receiveEvent, onReceive);
+            stream.removeListener('close', onClose);
+            stream.removeListener('error', onError);
+            stream.removeListener('end', onEnd);
+        }
+    });
+}
+
+module.exports = function (config) {
+    return function (stream, receiver, options) {
+        return read.call(this, stream, receiver, options, config);
+    };
+};