[Pkg-javascript-commits] [node-through] 01/02: Imported Upstream version 2.3.8
Thorsten Alteholz
alteholz at moszumanska.debian.org
Sun Mar 27 15:38:42 UTC 2016
This is an automated email from the git hooks/post-receive script.
alteholz pushed a commit to branch master
in repository node-through.
commit 040c32b13bc2e835051f8db31ae0bb4b40e0d379
Author: Thorsten Alteholz <debian at alteholz.de>
Date: Sun Mar 27 17:38:40 2016 +0200
Imported Upstream version 2.3.8
---
.travis.yml | 5 ++
LICENSE.APACHE2 | 15 ++++++
LICENSE.MIT | 24 ++++++++++
index.js | 108 +++++++++++++++++++++++++++++++++++++++++
package.json | 36 ++++++++++++++
readme.markdown | 64 +++++++++++++++++++++++++
test/async.js | 28 +++++++++++
test/auto-destroy.js | 30 ++++++++++++
test/buffering.js | 71 +++++++++++++++++++++++++++
test/end.js | 45 +++++++++++++++++
test/index.js | 133 +++++++++++++++++++++++++++++++++++++++++++++++++++
11 files changed, 559 insertions(+)
diff --git a/.travis.yml b/.travis.yml
new file mode 100644
index 0000000..c693a93
--- /dev/null
+++ b/.travis.yml
@@ -0,0 +1,5 @@
+language: node_js
+node_js:
+ - 0.6
+ - 0.8
+ - "0.10"
diff --git a/LICENSE.APACHE2 b/LICENSE.APACHE2
new file mode 100644
index 0000000..6366c04
--- /dev/null
+++ b/LICENSE.APACHE2
@@ -0,0 +1,15 @@
+Apache License, Version 2.0
+
+Copyright (c) 2011 Dominic Tarr
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
diff --git a/LICENSE.MIT b/LICENSE.MIT
new file mode 100644
index 0000000..6eafbd7
--- /dev/null
+++ b/LICENSE.MIT
@@ -0,0 +1,24 @@
+The MIT License
+
+Copyright (c) 2011 Dominic Tarr
+
+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.
diff --git a/index.js b/index.js
new file mode 100644
index 0000000..ca5fc59
--- /dev/null
+++ b/index.js
@@ -0,0 +1,108 @@
+var Stream = require('stream')
+
+// through
+//
+// a stream that does nothing but re-emit the input.
+// useful for aggregating a series of changing but not ending streams into one stream)
+
+exports = module.exports = through
+through.through = through
+
+//create a readable writable stream.
+
+function through (write, end, opts) {
+ write = write || function (data) { this.queue(data) }
+ end = end || function () { this.queue(null) }
+
+ var ended = false, destroyed = false, buffer = [], _ended = false
+ var stream = new Stream()
+ stream.readable = stream.writable = true
+ stream.paused = false
+
+// stream.autoPause = !(opts && opts.autoPause === false)
+ stream.autoDestroy = !(opts && opts.autoDestroy === false)
+
+ stream.write = function (data) {
+ write.call(this, data)
+ return !stream.paused
+ }
+
+ function drain() {
+ while(buffer.length && !stream.paused) {
+ var data = buffer.shift()
+ if(null === data)
+ return stream.emit('end')
+ else
+ stream.emit('data', data)
+ }
+ }
+
+ stream.queue = stream.push = function (data) {
+// console.error(ended)
+ if(_ended) return stream
+ if(data === null) _ended = true
+ buffer.push(data)
+ drain()
+ return stream
+ }
+
+ //this will be registered as the first 'end' listener
+ //must call destroy next tick, to make sure we're after any
+ //stream piped from here.
+ //this is only a problem if end is not emitted synchronously.
+ //a nicer way to do this is to make sure this is the last listener for 'end'
+
+ stream.on('end', function () {
+ stream.readable = false
+ if(!stream.writable && stream.autoDestroy)
+ process.nextTick(function () {
+ stream.destroy()
+ })
+ })
+
+ function _end () {
+ stream.writable = false
+ end.call(stream)
+ if(!stream.readable && stream.autoDestroy)
+ stream.destroy()
+ }
+
+ stream.end = function (data) {
+ if(ended) return
+ ended = true
+ if(arguments.length) stream.write(data)
+ _end() // will emit or queue
+ return stream
+ }
+
+ stream.destroy = function () {
+ if(destroyed) return
+ destroyed = true
+ ended = true
+ buffer.length = 0
+ stream.writable = stream.readable = false
+ stream.emit('close')
+ return stream
+ }
+
+ stream.pause = function () {
+ if(stream.paused) return
+ stream.paused = true
+ return stream
+ }
+
+ stream.resume = function () {
+ if(stream.paused) {
+ stream.paused = false
+ stream.emit('resume')
+ }
+ drain()
+ //may have become paused again,
+ //as drain emits 'data'.
+ if(!stream.paused)
+ stream.emit('drain')
+ return stream
+ }
+ return stream
+}
+
diff --git a/package.json b/package.json
new file mode 100644
index 0000000..9862189
--- /dev/null
+++ b/package.json
@@ -0,0 +1,36 @@
+{
+ "name": "through",
+ "version": "2.3.8",
+ "description": "simplified stream construction",
+ "main": "index.js",
+ "scripts": {
+ "test": "set -e; for t in test/*.js; do node $t; done"
+ },
+ "devDependencies": {
+ "stream-spec": "~0.3.5",
+ "tape": "~2.3.2",
+ "from": "~0.1.3"
+ },
+ "keywords": [
+ "stream",
+ "streams",
+ "user-streams",
+ "pipe"
+ ],
+ "author": "Dominic Tarr <dominic.tarr at gmail.com> (dominictarr.com)",
+ "license": "MIT",
+ "repository": {
+ "type": "git",
+ "url": "https://github.com/dominictarr/through.git"
+ },
+ "homepage": "https://github.com/dominictarr/through",
+ "testling": {
+ "browsers": [
+ "ie/8..latest",
+ "ff/15..latest",
+ "chrome/20..latest",
+ "safari/5.1..latest"
+ ],
+ "files": "test/*.js"
+ }
+}
diff --git a/readme.markdown b/readme.markdown
new file mode 100644
index 0000000..cb34c81
--- /dev/null
+++ b/readme.markdown
@@ -0,0 +1,64 @@
+#through
+
+[](http://travis-ci.org/dominictarr/through)
+[](https://ci.testling.com/dominictarr/through)
+
+Easy way to create a `Stream` that is both `readable` and `writable`.
+
+* Pass in optional `write` and `end` methods.
+* `through` takes care of pause/resume logic if you use `this.queue(data)` instead of `this.emit('data', data)`.
+* Use `this.pause()` and `this.resume()` to manage flow.
+* Check `this.paused` to see current flow state. (`write` always returns `!this.paused`).
+
+This function is the basis for most of the synchronous streams in
+[event-stream](http://github.com/dominictarr/event-stream).
+
+``` js
+var through = require('through')
+
+through(function write(data) {
+ this.queue(data) //data *must* not be null
+ },
+ function end () { //optional
+ this.queue(null)
+ })
+```
+
+Or, can also be used _without_ buffering on pause, use `this.emit('data', data)`,
+and this.emit('end')
+
+``` js
+var through = require('through')
+
+through(function write(data) {
+ this.emit('data', data)
+ //this.pause()
+ },
+ function end () { //optional
+ this.emit('end')
+ })
+```
+
+## Extended Options
+
+You will probably not need these 99% of the time.
+
+### autoDestroy=false
+
+By default, `through` emits close when the writable
+and readable side of the stream has ended.
+If that is not desired, set `autoDestroy=false`.
+
+``` js
+var through = require('through')
+
+//like this
+var ts = through(write, end, {autoDestroy: false})
+//or like this
+var ts = through(write, end)
+ts.autoDestroy = false
+```
+
+## License
+
+MIT / Apache2
diff --git a/test/async.js b/test/async.js
new file mode 100644
index 0000000..46bdbae
--- /dev/null
+++ b/test/async.js
@@ -0,0 +1,28 @@
+var from = require('from')
+var through = require('../')
+
+var tape = require('tape')
+
+tape('simple async example', function (t) {
+
+ var n = 0, expected = [1,2,3,4,5], actual = []
+ from(expected)
+ .pipe(through(function(data) {
+ this.pause()
+ n ++
+ setTimeout(function(){
+ console.log('pushing data', data)
+ this.push(data)
+ this.resume()
+ }.bind(this), 300)
+ })).pipe(through(function(data) {
+ console.log('pushing data second time', data);
+ this.push(data)
+ })).on('data', function (d) {
+ actual.push(d)
+ }).on('end', function() {
+ t.deepEqual(actual, expected)
+ t.end()
+ })
+
+})
diff --git a/test/auto-destroy.js b/test/auto-destroy.js
new file mode 100644
index 0000000..9a8fd00
--- /dev/null
+++ b/test/auto-destroy.js
@@ -0,0 +1,30 @@
+var test = require('tape')
+var through = require('../')
+
+// must emit end before close.
+
+test('end before close', function (assert) {
+ var ts = through()
+ ts.autoDestroy = false
+ var ended = false, closed = false
+
+ ts.on('end', function () {
+ assert.ok(!closed)
+ ended = true
+ })
+ ts.on('close', function () {
+ assert.ok(ended)
+ closed = true
+ })
+
+ ts.write(1)
+ ts.write(2)
+ ts.write(3)
+ ts.end()
+ assert.ok(ended)
+ assert.notOk(closed)
+ ts.destroy()
+ assert.ok(closed)
+ assert.end()
+})
+
diff --git a/test/buffering.js b/test/buffering.js
new file mode 100644
index 0000000..b0084bf
--- /dev/null
+++ b/test/buffering.js
@@ -0,0 +1,71 @@
+var test = require('tape')
+var through = require('../')
+
+// must emit end before close.
+
+test('buffering', function(assert) {
+ var ts = through(function (data) {
+ this.queue(data)
+ }, function () {
+ this.queue(null)
+ })
+
+ var ended = false, actual = []
+
+ ts.on('data', actual.push.bind(actual))
+ ts.on('end', function () {
+ ended = true
+ })
+
+ ts.write(1)
+ ts.write(2)
+ ts.write(3)
+ assert.deepEqual(actual, [1, 2, 3])
+ ts.pause()
+ ts.write(4)
+ ts.write(5)
+ ts.write(6)
+ assert.deepEqual(actual, [1, 2, 3])
+ ts.resume()
+ assert.deepEqual(actual, [1, 2, 3, 4, 5, 6])
+ ts.pause()
+ ts.end()
+ assert.ok(!ended)
+ ts.resume()
+ assert.ok(ended)
+ assert.end()
+})
+
+test('buffering has data in queue, when ends', function (assert) {
+
+ /*
+ * If stream ends while paused with data in the queue,
+ * stream should still emit end after all data is written
+ * on resume.
+ */
+
+ var ts = through(function (data) {
+ this.queue(data)
+ }, function () {
+ this.queue(null)
+ })
+
+ var ended = false, actual = []
+
+ ts.on('data', actual.push.bind(actual))
+ ts.on('end', function () {
+ ended = true
+ })
+
+ ts.pause()
+ ts.write(1)
+ ts.write(2)
+ ts.write(3)
+ ts.end()
+ assert.deepEqual(actual, [], 'no data written yet, still paused')
+ assert.ok(!ended, 'end not emitted yet, still paused')
+ ts.resume()
+ assert.deepEqual(actual, [1, 2, 3], 'resumed, all data should be delivered')
+ assert.ok(ended, 'end should be emitted once all data was delivered')
+ assert.end();
+})
diff --git a/test/end.js b/test/end.js
new file mode 100644
index 0000000..fa113f5
--- /dev/null
+++ b/test/end.js
@@ -0,0 +1,45 @@
+var test = require('tape')
+var through = require('../')
+
+// must emit end before close.
+
+test('end before close', function (assert) {
+ var ts = through()
+ var ended = false, closed = false
+
+ ts.on('end', function () {
+ assert.ok(!closed)
+ ended = true
+ })
+ ts.on('close', function () {
+ assert.ok(ended)
+ closed = true
+ })
+
+ ts.write(1)
+ ts.write(2)
+ ts.write(3)
+ ts.end()
+ assert.ok(ended)
+ assert.ok(closed)
+ assert.end()
+})
+
+test('end only once', function (t) {
+
+ var ts = through()
+ var ended = false, closed = false
+
+ ts.on('end', function () {
+ t.equal(ended, false)
+ ended = true
+ })
+
+ ts.queue(null)
+ ts.queue(null)
+ ts.queue(null)
+
+ ts.resume()
+
+ t.end()
+})
diff --git a/test/index.js b/test/index.js
new file mode 100644
index 0000000..96da82f
--- /dev/null
+++ b/test/index.js
@@ -0,0 +1,133 @@
+
+var test = require('tape')
+var spec = require('stream-spec')
+var through = require('../')
+
+/*
+ I'm using these two functions, and not streams and pipe
+ so there is less to break. if this test fails it must be
+ the implementation of _through_
+*/
+
+function write(array, stream) {
+ array = array.slice()
+ function next() {
+ while(array.length)
+ if(stream.write(array.shift()) === false)
+ return stream.once('drain', next)
+
+ stream.end()
+ }
+
+ next()
+}
+
+function read(stream, callback) {
+ var actual = []
+ stream.on('data', function (data) {
+ actual.push(data)
+ })
+ stream.once('end', function () {
+ callback(null, actual)
+ })
+ stream.once('error', function (err) {
+ callback(err)
+ })
+}
+
+test('simple defaults', function(assert) {
+
+ var l = 1000
+ , expected = []
+
+ while(l--) expected.push(l * Math.random())
+
+ var t = through()
+ var s = spec(t).through().pausable()
+
+ read(t, function (err, actual) {
+ assert.ifError(err)
+ assert.deepEqual(actual, expected)
+ assert.end()
+ })
+
+ t.on('close', s.validate)
+
+ write(expected, t)
+});
+
+test('simple functions', function(assert) {
+
+ var l = 1000
+ , expected = []
+
+ while(l--) expected.push(l * Math.random())
+
+ var t = through(function (data) {
+ this.emit('data', data*2)
+ })
+ var s = spec(t).through().pausable()
+
+
+ read(t, function (err, actual) {
+ assert.ifError(err)
+ assert.deepEqual(actual, expected.map(function (data) {
+ return data*2
+ }))
+ assert.end()
+ })
+
+ t.on('close', s.validate)
+
+ write(expected, t)
+})
+
+test('pauses', function(assert) {
+
+ var l = 1000
+ , expected = []
+
+ while(l--) expected.push(l) //Math.random())
+
+ var t = through()
+
+ var s = spec(t)
+ .through()
+ .pausable()
+
+ t.on('data', function () {
+ if(Math.random() > 0.1) return
+ t.pause()
+ process.nextTick(function () {
+ t.resume()
+ })
+ })
+
+ read(t, function (err, actual) {
+ assert.ifError(err)
+ assert.deepEqual(actual, expected)
+ })
+
+ t.on('close', function () {
+ s.validate()
+ assert.end()
+ })
+
+ write(expected, t)
+})
+
+test('does not soft-end on `undefined`', function(assert) {
+ var stream = through()
+ , count = 0
+
+ stream.on('data', function (data) {
+ count++
+ })
+
+ stream.write(undefined)
+ stream.write(undefined)
+
+ assert.equal(count, 2)
+
+ assert.end()
+})
--
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/pkg-javascript/node-through.git
More information about the Pkg-javascript-commits
mailing list