流入门

readable._read(size)

readable.push(chunk)

import { Readable } from 'stream'
import Chance from 'chance'

const chance = new Chance()

export class RandomStream extends Readable {
    constructor (options) {
        super(options)
        this.emittedBytes = 0
    }

    _read (size) {
        const chunk = chance.string({ length: size }) // (1)
        this.push(chunk, 'utf8') // (2)
        this.emittedBytes += chunk.length
        if (chance.bool({ likelihood: 5 })) { // (3)
            this.push(null)
        }
    }
}

// index.js
import { RandomStream } from './random-stream.js'

const randomStream = new RandomStream()
randomStream
    .on('data', (chunk) => {
        console.log(`Chunk received (${chunk.length} bytes): ${chunk.toString()}`)
    })
    .on('end', () => {
        console.log(`Produced ${randomStream.emittedBytes} bytes of random data`)
    })

writable.write(chunk, [encoding], [callback])

writable.end([chunk], [encoding], [callback])

// entropy-server.js
import { createServer } from 'http'
import Chance from 'chance'

const chance = new Chance()
const server = createServer((req, res) => {
    res.writeHead(200, { 'Content-Type': 'text/plain' }) // (1)
    while (chance.bool({ likelihood: 95 })) { // (2)
        res.write(`${chance.string()}\n`) // (3)
    }
    res.end('\n\n') // (4)
    res.on('finish', () => console.log('All data sent')) // (5)
})
server.listen(8080, () => {
    console.log('listening on http://localhost:8080')
})

curl localhost:8080

// ...
const server = createServer((req, res) => {
    res.writeHead(200, { 'Content-Type': 'text/plain' })
    function generateMore () { // (1)
        while (chance.bool({ likelihood: 95 })) {
            const randomChunk = chance.string({ // (2)
                length: (16 * 1024) - 1
            })
            const shouldContinue = res.write(`${randomChunk}\n`) // (3)
            if (!shouldContinue) {
                console.log('back-pressure')
                return res.once('drain', generateMore)
            }
        }
        res.end('\n\n')
    }
    generateMore()
    res.on('finish', () => console.log('All data sent'))
})
// ...

{
    path: <path to a file>
    content: <string or buffer>
}

import { Writable } from 'stream'
import { promises as fs } from 'fs'
import { dirname } from 'path'
import mkdirp from 'mkdirp-promise'

export class ToFileStream extends Writable {
    constructor (options) {
        super({ ...options, objectMode: true })
    }

    _write (chunk, encoding, cb) {
        mkdirp(dirname(chunk.path))
            .then(() => fs.writeFile(chunk.path, chunk.content))
            .then(() => cb())
            .catch(cb)
    }
}

import { join } from 'path'
import { ToFileStream } from './to-file-stream.js'
const tfs = new ToFileStream()

tfs.write({
    path: join('files', 'file1.txt'), content: 'Hello' })
tfs.write({
    path: join('files', 'file2.txt'), content: 'Node.js' })
tfs.write({
    path: join('files', 'file3.txt'), content: 'streams' })
tfs.end(() => console.log('All files created'))

import { Transform } from 'stream'

export class ReplaceStream extends Transform {
    constructor (searchStr, replaceStr, options) {
        super({ ...options })
        this.searchStr = searchStr
        this.replaceStr = replaceStr
        this.tail = ''
    }

    _transform (chunk, encoding, callback) {
        const pieces = (this.tail + chunk).split(this.searchStr) // (1)
        const lastPiece = pieces[pieces.length - 1] // (2)
        const tailLen = this.searchStr.length - 1
        this.tail = lastPiece.slice(-tailLen)
        pieces[pieces.length - 1] = lastPiece.slice(0, -tailLen)
        this.push(pieces.join(this.replaceStr)) // (3)
        callback()
    }

    _flush (callback) {
        this.push(this.tail)
        callback()
    }
}

import { ReplaceStream } from './replace-stream.js'

const replaceStream = new ReplaceStream('World', 'Node.js')
replaceStream.on('data', chunk => console.log(chunk.toString()))

replaceStream.write('Hello W')
replaceStream.write('orld!')
replaceStream.end()

Hel
lo Node.js
!

type,country,profit
Household,Namibia,597290.92
Baby Food,Iceland,808579.10
Meat,Russia,277305.60
Meat,Italy,413270.00
Cereal,Malta,174965.25
Meat,Indonesia,145402.40
Household,Italy,728880.54
[... many more lines]

import { createReadStream } from 'fs'
import parse from 'csv-parse'
import { FilterByCountry } from './filter-by-country.js'
import { SumProfit } from './sum-profit.js'

const csvParser = parse({ columns: true })

createReadStream('data.csv') // (1)
    .pipe(csvParser) // (2)
    .pipe(new FilterByCountry('Italy')) // (3)
    .pipe(new SumProfit()) // (4)
    .pipe(process.stdout) // (5)

import { Transform } from 'stream'

export class FilterByCountry extends Transform {
    constructor (country, options = {}) {
        options.objectMode = true
        super(options)
        this.country = country
    }

    _transform (record, enc, cb) {
        if (record.country === this.country) {
            this.push(record)
        }
        cb()
    }
}

import { Transform } from 'stream'

export class SumProfit extends Transform {
    constructor (options = {}) {
        options.objectMode = true
        super(options)
        this.total = 0
    }

    _transform (record, enc, cb) {
        this.total += Number.parseFloat(record.profit)
        cb()
    }

    _flush (cb) {
        this.push(this.total.toString())
        cb()
    }
}

import { PassThrough } from 'stream'

let bytesWritten = 0
const monitor = new PassThrough()
monitor.on('data', (chunk) => {
    bytesWritten += chunk.length
})
monitor.on('finish', () => {
    console.log(`${bytesWritten} bytes written`)
})

monitor.write('Hello!')
monitor.end()

createReadStream(filename)
    .pipe(createGzip())
    .pipe(monitor)
    .pipe(createWriteStream(`${filename}.gz`))

function upload (filename, contentStream) {
    // ...
}

import { createReadStream } from 'fs'
upload('a-picture.jpg', createReadStream('/path/to/a-picture.jpg'))

import { createReadStream } from 'fs'
import { createBrotliCompress } from 'zlib'
import { PassThrough } from 'stream'
import { basename } from 'path'
import { upload } from './upload.js'

const filepath = process.argv[2] // (1)
const filename = basename(filepath)
const contentStream = new PassThrough() // (2)

upload(`${filename}.br`, contentStream) // (3)
    .then((response) => {
        console.log(`Server response: ${response.data}`)
    })
    .catch((err) => {
        console.error(err)
        process.exit(1)
    })

createReadStream(filepath) // (4)
    .pipe(createBrotliCompress())
    .pipe(contentStream)

function createUploadStream (filename) {
    // ...
    // returns a writable stream that can be used to upload data
}

function createUploadStream (filename) {
    const connector = new PassThrough()
    upload(filename, connector)
    return connector
}

const upload = createUploadStream('a-file.txt')
upload.write('Hello World')
upload.end()

stream1
    .pipe(stream2)
    .on('error', () => {})

stream1
    .on('error', () => {})
    .pipe(stream2)
    .on('error', () => {})

function handleError (err) {
    console.error(err)
    stream1.destroy()
    stream2.destroy()
}
stream1
    .on('error', handleError)
    .pipe(stream2)
    .on('error', handleError)

pipeline(stream1, stream2, stream3, ... , cb)

import { createGzip, createGunzip } from 'zlib' // (1)
import { Transform, pipeline } from 'stream'

const uppercasify = new Transform({ // (2)
    transform (chunk, enc, cb) {
        this.push(chunk.toString().toUpperCase())
        cb()
    }
})

pipeline( // (3)
    process.stdin,
    createGunzip(),
    uppercasify,
    createGzip(),
    process.stdout,
    (err) => { // (4)
        if (err) {
            console.error(err)
            process.exit(1)
        }
    }
)

echo 'Hello World!' | gzip | node uppercasify-gzipped.js | gunzip

HELLO WORLD!

Error: unexpected end of file
at Zlib.zlibOnError [as onerror] (zlib.js:180:17) {
errno: -5,
code: 'Z_BUF_ERROR'
}