Firebase 函数 Node.js 转换流

Posted 2023-02-22

技术标签:

【中文标题】Firebase 函数 Node.js 转换流【英文标题】：Firebase function Node.js transform stream 【发布时间】：2021-01-22 12:50:29 【问题描述】：

我正在创建一个 Firebase HTTP 函数，它进行 BigQuery 查询并返回查询结果的修改版本。该查询可能会返回数百万行，因此在响应 HTTP 客户端之前，我无法将整个查询结果存储在内存中。我正在尝试使用 Node.js 流，并且由于我需要在将结果发送到客户端之前对其进行修改，因此我正在尝试使用转换流。但是，当我尝试通过我的转换流传输查询流时，Firebase 函数崩溃并显示以下错误消息：finished with status: 'response error'。

我的最小可重现示例如下。我正在使用缓冲区，因为我不想一次处理单行（块），因为我需要进行异步网络调用来转换数据。

return new Promise((resolve, reject) => 
    const buffer = new Array(5000)
    let bufferIndex = 0
    const [job] = await bigQuery.createQueryJob(options)
    const bqStream = job.getQueryResultsStream()

    const transformer = new Transform(
        writableObjectMode: true,
        readableObjectMode: false,
        transform(chunk, enc, callback) 
            buffer[bufferIndex] = chunk
            if (bufferIndex < buffer.length - 1) 
                bufferIndex++
            
            else 
                this.push(JSON.stringify(buffer).slice(1, -1)) // Transformation should happen here.
                bufferIndex = 0
            
            callback()
        ,
        flush(callback) 
            if (bufferIndex > 0) 
                this.push(JSON.stringify(buffer.slice(0, bufferIndex)).slice(1, -1))
            
            this.push("]")
            callback()
        ,
    )

    bqStream
        .pipe(transform)
        .pipe(response)

    bqStream.on("end", () => 
        resolve()
    )

【问题讨论】：

【参考方案1】：