var tape = require('tape')
|
var through = require('through2')
|
var pumpify = require('./')
|
var stream = require('stream')
|
var duplexify = require('duplexify')
|
|
tape('basic', function(t) {
|
t.plan(3)
|
|
var pipeline = pumpify(
|
through(function(data, enc, cb) {
|
t.same(data.toString(), 'hello')
|
cb(null, data.toString().toUpperCase())
|
}),
|
through(function(data, enc, cb) {
|
t.same(data.toString(), 'HELLO')
|
cb(null, data.toString().toLowerCase())
|
})
|
)
|
|
pipeline.write('hello')
|
pipeline.on('data', function(data) {
|
t.same(data.toString(), 'hello')
|
t.end()
|
})
|
})
|
|
tape('3 times', function(t) {
|
t.plan(4)
|
|
var pipeline = pumpify(
|
through(function(data, enc, cb) {
|
t.same(data.toString(), 'hello')
|
cb(null, data.toString().toUpperCase())
|
}),
|
through(function(data, enc, cb) {
|
t.same(data.toString(), 'HELLO')
|
cb(null, data.toString().toLowerCase())
|
}),
|
through(function(data, enc, cb) {
|
t.same(data.toString(), 'hello')
|
cb(null, data.toString().toUpperCase())
|
})
|
)
|
|
pipeline.write('hello')
|
pipeline.on('data', function(data) {
|
t.same(data.toString(), 'HELLO')
|
t.end()
|
})
|
})
|
|
tape('destroy', function(t) {
|
var test = through()
|
test.destroy = function() {
|
t.ok(true)
|
t.end()
|
}
|
|
var pipeline = pumpify(through(), test)
|
|
pipeline.destroy()
|
})
|
|
tape('close', function(t) {
|
var test = through()
|
var pipeline = pumpify(through(), test)
|
|
pipeline.on('error', function(err) {
|
t.same(err.message, 'lol')
|
t.end()
|
})
|
|
test.emit('error', new Error('lol'))
|
})
|
|
tape('end waits for last one', function(t) {
|
var ran = false
|
|
var a = through()
|
var b = through()
|
var c = through(function(data, enc, cb) {
|
setTimeout(function() {
|
ran = true
|
cb()
|
}, 100)
|
})
|
|
var pipeline = pumpify(a, b, c)
|
|
pipeline.write('foo')
|
pipeline.end(function() {
|
t.ok(ran)
|
t.end()
|
})
|
|
t.ok(!ran)
|
})
|
|
tape('always wait for finish', function(t) {
|
var a = new stream.Readable()
|
a._read = function() {}
|
a.push('hello')
|
|
var pipeline = pumpify(a, through(), through())
|
var ran = false
|
|
pipeline.on('finish', function() {
|
t.ok(ran)
|
t.end()
|
})
|
|
setTimeout(function() {
|
ran = true
|
a.push(null)
|
}, 100)
|
})
|
|
tape('async', function(t) {
|
var pipeline = pumpify()
|
|
t.plan(4)
|
|
pipeline.write('hello')
|
pipeline.on('data', function(data) {
|
t.same(data.toString(), 'HELLO')
|
t.end()
|
})
|
|
setTimeout(function() {
|
pipeline.setPipeline(
|
through(function(data, enc, cb) {
|
t.same(data.toString(), 'hello')
|
cb(null, data.toString().toUpperCase())
|
}),
|
through(function(data, enc, cb) {
|
t.same(data.toString(), 'HELLO')
|
cb(null, data.toString().toLowerCase())
|
}),
|
through(function(data, enc, cb) {
|
t.same(data.toString(), 'hello')
|
cb(null, data.toString().toUpperCase())
|
})
|
)
|
}, 100)
|
})
|
|
tape('early destroy', function(t) {
|
var a = through()
|
var b = through()
|
var c = through()
|
|
b.destroy = function() {
|
t.ok(true)
|
t.end()
|
}
|
|
var pipeline = pumpify()
|
|
pipeline.destroy()
|
setTimeout(function() {
|
pipeline.setPipeline(a, b, c)
|
}, 100)
|
})
|
|
tape('preserves error', function (t) {
|
var a = through()
|
var b = through(function (data, enc, cb) {
|
cb(new Error('stop'))
|
})
|
var c = through()
|
var s = pumpify()
|
|
s.on('error', function (err) {
|
t.same(err.message, 'stop')
|
t.end()
|
})
|
|
s.setPipeline(a, b, c)
|
s.resume()
|
s.write('hi')
|
})
|
|
tape('preserves error again', function (t) {
|
var ws = new stream.Writable()
|
var rs = new stream.Readable({highWaterMark: 16})
|
|
ws._write = function (data, enc, cb) {
|
cb(null)
|
}
|
|
rs._read = function () {
|
process.nextTick(function () {
|
rs.push('hello world')
|
})
|
}
|
|
var pumpifyErr = pumpify(
|
through(),
|
through(function(chunk, _, cb) {
|
cb(new Error('test'))
|
}),
|
ws
|
)
|
|
rs.pipe(pumpifyErr)
|
.on('error', function (err) {
|
t.ok(err)
|
t.ok(err.message !== 'premature close', 'does not close with premature close')
|
t.end()
|
})
|
})
|
|
tape('returns error from duplexify', function (t) {
|
var a = through()
|
var b = duplexify()
|
var s = pumpify()
|
|
s.setPipeline(a, b)
|
|
s.on('error', function (err) {
|
t.same(err.message, 'stop')
|
t.end()
|
})
|
|
s.write('data')
|
// Test passes if `.end()` is not called
|
s.end()
|
|
b.setWritable(through())
|
|
setImmediate(function () {
|
b.destroy(new Error('stop'))
|
})
|
})
|
