Stream | Description |
---|---|
Readable |
Data emitter |
Writable |
Data receiver |
Transform |
Emitter and receiver |
Duplex |
Emitter and receiver (independent) |
See: Stream (nodejs.org)
const Readable = require('stream').Readable
const Writable = require('stream').Writable
const Transform = require('stream').Transform
clock() // Readable stream
.pipe(xformer()) // Transform stream
.pipe(renderer()) // Writable stream
stream.push(/*...*/) // Emit a chunk
stream.emit('error', error) // Raise an error
stream.push(null) // Close a stream
const st = source()
st.on('data', (data) => { console.log('<-', data) })
st.on('error', (err) => { console.log('!', err.message) })
st.on('close', () => { console.log('** bye') })
st.on('finish', () => { console.log('** bye') })
Assuming source()
is a readable stream.
// Toggle flowing mode
st.resume()
st.pause()
// Automatically turns on flowing mode
st.on('data', /*...*/)
{: .-three-column}
function clock () {
const stream = new Readable({
objectMode: true,
read() {}
})
setInterval(() => {
stream.push({ time: new Date() })
}, 1000)
return stream
}
// Implement read() if you
// need on-demand reading.
Readable streams are generators of data. Write data using stream.push()
.
function xformer () {
let count = 0
return new Transform({
objectMode: true,
transform: (data, _, done) => {
done(null, { ...data, index: count++ })
}
})
}
Pass the updated chunk to done(null, chunk)
.
function renderer () {
return new Writable({
objectMode: true,
write: (data, _, done) => {
console.log('<-', data)
done()
}
})
}
clock() // Readable stream
.pipe(xformer()) // Transform stream
.pipe(renderer()) // Writable stream
{: .-one-column}
Subscribe to get resources directly to your inbox. You won't receive any spam! ✌️