在节点的双工流中读取和处理较小的块答案

【问题标题】：Read and process smaller chunks inside a duplex stream in node在节点的双工流中读取和处理较小的块
【发布时间】：2016-03-27 14:12:30
【问题描述】：

我需要获取一个 20Mb json 对象数组的序列，通过管道传输到流中，分成 33 个项目的较小数组，将其转换为 html，然后通过管道传输到另一个流（用于 pdf 转换）。

问题是，我还不太了解节点流的工作原理。我正在尝试使用双工流解决它，但我不知道如何汇集来自上游的传入块并将它们部分发送到下游。在这段代码中

jsonReader = fs.createReadStream 'source.json'

class Convert extends Duplex

    constructor: ->
        super readableObjectMode: true
        # Duplex.call @, readableObjectMode: true
        @buffer = []

    _read: (lines) ->
        console.log "buffer #{@buffer.length}"
        if @buffer.length is 0
            @push null
        else 
            console.log "lines: #{lines}"
            page = @buffer.slice 0, 33
            console.log page.length
            @buffer.splice 0, 33
            @push page

    _write: (data, enconding, next) ->
        @buffer.push data
        next()

convert = new Convert()

jsonReader.pipe(convert).pipe(process.stdout)

@buffer 始终为空。节点将来自上游的块存储在哪里？

【问题讨论】：

标签： node.js coffeescript stream duplex

【解决方案1】：

您在_write 中接收的data 是一个缓冲区，是输入文件的二进制部分，您不会接收对象甚至字符串。您可以手动解析块以检索对象，也可以将整个文件加载到内存中（20Mb 不是那么大）并解析它。这是一个示例（我使用event-stream 来轻松操作/创建流）：

es = require('event-stream')

convert = (path) ->
    # load and parse your file
    content = JSON.parse fs.readFileSync(path, 'utf8')

    es.readable (count, next) ->
        # emit 33 elements at a time until content.length === 0
        while content.length
            this.emit 'data', content.splice(0, 33)

        # close the stream
        this.emit 'end'
        next()

srcPath = __dirname + '/source.json'
# convert is a stream
convert(srcPath)
    # piping to console.log because process.stdout can't handle objects
    .pipe(es.map (arr, next) ->
        console.log arr
        next()
    )

【讨论】：