init

.editorconfig

@@ -0,0 +1,12 @@
+root = true
+
+[*]
+charset = utf-8
+end_of_line = lf
+indent_style = space
+indent_size = 2
+insert_final_newline = true
+trim_trailing_whitespace = true
+
+[*.md]
+trim_trailing_whitespace = false

.gitattributes

@@ -0,0 +1 @@
+* text=auto

.gitignore

@@ -0,0 +1,2 @@
+coverage
+node_modules

.travis.yml

@@ -0,0 +1,8 @@
+dist: trusty
+branches:
+  except: /^v\d/
+language: node_js
+node_js: node
+after_script:
+  - npm install istanbul-coveralls
+  - node node_modules/.bin/istanbul-coveralls

LICENSE

@@ -0,0 +1,20 @@
+The MIT License (MIT)
+
+Copyright (c) 2017 Shinnosuke Watanabe
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,66 @@
+# cancelable-pump
+
+[![NPM version](https://img.shields.io/npm/v/cancelable-pump.svg)](https://www.npmjs.com/package/cancelable-pump)
+[![Build Status](https://travis-ci.org/shinnn/cancelable-pump.svg?branch=master)](https://travis-ci.org/shinnn/cancelable-pump)
+[![Coverage Status](https://img.shields.io/coveralls/shinnn/cancelable-pump.svg)](https://coveralls.io/github/shinnn/cancelable-pump?branch=master)
+
+Cancelable [`pump`](https://github.com/mafintosh/pump)
+
+```javascript
+const {createReadStream, createWriteStream} = require('cancelable-pump');
+const cancelablePump = require('cancelable-pump');
+
+cancelablePump(createReadStream('1GB-file.txt'), createWriteStream('dest0'), () => {
+  statSync('dest0').size; //=> 1000000000;
+});
+
+const cancel = cancelablePump(createReadStream('1GB-file.txt'), createWriteStream('dest1'), () => {
+  statSync('dest1').size; //=> 263192576, or something else smaller than 1000000000
+});
+
+setTimeout(() => {
+  cancel();
+}, 1000);
+```
+
+## Installation
+
+[Use npm.](https://docs.npmjs.com/cli/install)
+
+```
+npm install cancelable-pump
+```
+
+## API
+
+```javascript
+const cancelablePump = require('cancelable-pump');
+```
+
+### cancelablePump(*stream0* [, *stream1*, *stream2*, ...] [, *callback*])
+
+*stream0*, *stream1*, *stream2*, ...: [`Stream`](https://nodejs.org/api/stream.html#stream_stream)  
+*callback*: `Function`  
+Return: `Function`
+
+### cancelablePump(*streams* [, *callback*])
+
+*streams*: `Array` of `Stream`s  
+*callback*: `Function`  
+Return: `Function`
+
+The API is almost the same as `pump`'s. The only difference is *cancelable-pump* returns a function to destroy all streams without passing any errors to the callback.
+
+```javascript
+const cancel = cancelablePump([src, transform, anotherTransform, dest], err => {
+  err; //=> undefined
+});
+
+cancel();
+```
+
+## License
+
+Copyright (c) 2017 [Shinnosuke Watanabe](https://github.com/shinnn)
+
+Licensed under [the MIT License](./LICENSE).

index.js

@@ -0,0 +1,41 @@
+/*!
+ * cancelable-pump | MIT (c) Shinnosuke Watanabe
+ * https://github.com/shinnn/cancelable-pump
+*/
+'use strict';
+
+var pump = require('pump');
+
+var cancel = new Error('Canceled.');
+
+module.exports = function cancelablePump() {
+  var nonCallbackArgs = Array.prototype.slice.call(arguments); // eslint-disable-line prefer-rest-params
+  var callback = typeof nonCallbackArgs[nonCallbackArgs.length - 1] === 'function' ?
+                 nonCallbackArgs.pop() :
+                 null;
+
+  var streams = Array.isArray(nonCallbackArgs[0]) ? nonCallbackArgs[0] : nonCallbackArgs;
+  var streamCount = streams.length;
+
+  if (streamCount < 2) {
+    throw new RangeError('cancelable-pump requires more than 2 streams, but got ' + streamCount + '.');
+  }
+
+  pump(streams, function(err) {
+    if (!callback) {
+      return;
+    }
+
+    if (err && err !== cancel) {
+      callback(err);
+      return;
+    }
+
+    callback();
+  });
+
+  return function cancelStreams() {
+    streams[streamCount - 1].emit('error', cancel);
+  };
+};
+

package.json

@@ -0,0 +1,40 @@
+{
+  "name": "cancelable-pump",
+  "version": "0.1.0",
+  "description": "Cancelable `pump`",
+  "repository": "shinnn/cancelable-pump",
+  "author": "Shinnosuke Watanabe (https://github.com/shinnn)",
+  "scripts": {
+    "pretest": "eslint --fix --format=codeframe index.js test.js",
+    "test": "istanbul cover test.js"
+  },
+  "license": "MIT",
+  "files": [
+    "index.js"
+  ],
+  "keywords": [
+    "pump",
+    "stream",
+    "pipe",
+    "join",
+    "chain",
+    "cancel",
+    "cancellation",
+    "abort",
+    "stop",
+    "destroy",
+    "callback"
+  ],
+  "dependencies": {
+    "pump": "^1.0.2"
+  },
+  "devDependencies": {
+    "@shinnn/eslint-config-node-legacy": "^3.0.0",
+    "eslint": "^3.18.0",
+    "istanbul": "^0.4.5",
+    "tape": "^4.6.3"
+  },
+  "eslintConfig": {
+    "extends": "@shinnn/node-legacy"
+  }
+}

test.js

@@ -0,0 +1,56 @@
+'use strict';
+
+const {PassThrough} = require('stream');
+
+const cancelablePump = require('.');
+const test = require('tape');
+
+test('cancelablePump()', t => {
+  t.plan(3);
+
+  t.doesNotThrow(
+    () => cancelablePump([new PassThrough(), new PassThrough()])(),
+    'should return a function.'
+  );
+
+  const cancel = cancelablePump(new PassThrough(), new PassThrough(), err => {
+    t.strictEqual(
+      err,
+      undefined,
+      'should cancel all streams when the cancellation function is called.'
+    );
+  });
+
+  cancel();
+
+  const dest1 = new PassThrough();
+  const error = new Error();
+
+  cancelablePump(new PassThrough(), dest1, err => {
+    t.strictEqual(err, error, 'should pass normal errors as it is.');
+  });
+
+  dest1.emit('error', error);
+});
+
+test('Argument validation', t => {
+  t.throws(
+    () => cancelablePump(),
+    /^RangeError: cancelable-pump requires more than 2 streams, but got 0\.$/,
+    'should throw an error when it takes no arguments.'
+  );
+
+  t.throws(
+    () => cancelablePump(new PassThrough()),
+    /^RangeError: cancelable-pump requires more than 2 streams, but got 1\.$/,
+    'should throw an error when it takes only 1 argument.'
+  );
+
+  t.throws(
+    () => cancelablePump(new PassThrough(), t.fail),
+    /^RangeError: cancelable-pump requires more than 2 streams, but got 1\.$/,
+    'should throw an error when the arguments include less than 2 streams.'
+  );
+
+  t.end();
+});