node.js - Node 转换流未将输出发送到管道？

Question

我正在做什么

我正在尝试编写一个转换流，该流将用作从 HTTP API (Salesforce) 读取数据的过程的一部分，将每个资源转换为标准内部数据结构，然后发送转换后的数据以供 Elasticsearch 编制索引。

我的问题

我编写了一个简单的转换流，但它似乎正在吞噬数据，而不是传递数据。

这是我的类AdapterStream

const { Transform } = require('stream')

class AdapterStream extends Transform {
  constructor (adapter) {
    super({objectMode: true})
    // adapter is a function that contains the transform logic
    this.adapter = adapter
  }

  _transform (chunk, _encoding, callback) {
    this.push(this.adapter(chunk))
    callback()
  }
}

module.exports = AdapterStream

这是我的使用方法

// query is an EventEmitter (https://jsforce.github.io/document/#query)
query.run({autoFetch: true, maxFetch: 10})
     .pipe(transform)
     .pipe(process.stdout)

我期望发生什么

我希望在终端窗口中看到转换后的数据。

实际发生的情况

我的终端窗口中没有打印任何内容

我尝试过的事情

写入文件

query.run({autoFetch: true, maxFetch: 10})
     .pipe(transform)
     .pipe(fs.createWriteStream('./test.json'))

这将按预期创建文件 ./test.json，但它是空的。

订阅事件

query.run({autoFetch: true, maxFetch: 10})
     .pipe(transform)

transform.on('readable', () => { console.log(transform.read()) })

这完成了我期望的 .pipe(process.stdout) 操作:将转换后的记录打印到控制台。

我的问题是，为什么 pipe() 没有执行我期望的操作？一定有一些简单的东西我错过了。

Answer 1

您希望将数据作为第二个参数返回给回调。就像下面这样。

const { Transform } = require('stream')

class AdapterStream extends Transform {
  constructor (adapter) {
    super({objectMode: true})
    // adapter is a function that contains the transform logic
    this.adapter = adapter
  }

  transform (chunk, _encoding, callback) {
    callback(null, this.adapter(chunk))
  }
}