如何将可写缓冲区通过管道传输到 ReadStream？答案

【问题标题】：How to pipe Writable Buffer to a ReadStream?如何将可写缓冲区通过管道传输到 ReadStream？
【发布时间】：2019-09-12 19:44:14
【问题描述】：

如何获取可写流并从缓冲区返回可读流？

我有以下内容可以将来自 ftp 服务器的数据写入块数组：

let chunks = []
let writable = new Writable
writable._write = (chunk, encoding, callback) => {
  chunks.push(chunk)
  callback()
}

然后我正在创建一个新的读取流：

let readable = new ReadStream()

然后我尝试将可写的管道传输到可读的管道，但这似乎不起作用：

“ReadStream”类型的参数不能分配给“WritableStream”类型的参数。

writable.pipe(readable)

这是整个方法：

export class FTP {
  readStream(filePath, options = {}) {
    let conn = this.getConnection(this.name)
    if (!conn) return Buffer.from('')
    filePath = this.forceRoot(filePath) 
    let chunks = []
    let writable = new Writable
    writable._write = (chunk, encoding, callback) => {
      chunks.push(chunk)
      callback()
    }
    let readable = new ReadStream()

    conn.client.download(writable, filePath, options.start || undefined)
    writable.pipe(readable)
    return readable
  }
}

然后我从流中读取并将输出通过管道传输到从http.createServer() 创建的响应对象，如下所示：

      let stream = store.readStream(file, { start, end })
        .on('open', () => stream.pipe(res))
        .on('close', () => res.end())
        .on('error', err => res.end(err))

【问题讨论】：

标签： javascript node.js

【解决方案1】：

是的，Node.js 流很难掌握。从逻辑上讲，这里不需要两个流。如果您想像从流中一样从 FTP 类中读取，您只需要实现单个可读流。查看文档中的this section，了解如何从头开始实现可读流：

class SourceWrapper extends Readable {
  constructor(options) {
    super(options);

    this._source = getLowLevelSourceObject();

    // Every time there's data, push it into the internal buffer.
    this._source.ondata = (chunk) => {
      // If push() returns false, then stop reading from source.
      if (!this.push(chunk))
        this._source.readStop();
    };

    // When the source ends, push the EOF-signaling `null` chunk.
    this._source.onend = () => {
      this.push(null);
    };
  }
  // _read() will be called when the stream wants to pull more data in.
  // The advisory size argument is ignored in this case.
  _read(size) {
    this._source.readStart();
  }
}

但是，从您的示例中，我可以得出结论，conn.client.download() 需要一个可写流作为输入参数。在这种情况下，您只需采用标准的PassThrough 流，它是一个双工（即左侧可写，右侧可读取）流，不应用任何转换：

const { PassThrough } = require('stream');

export class FTP {
  readStream(filePath, options = {}) {
    let conn = this.getConnection(this.name);
    if (!conn) return Buffer.from('');
    filePath = this.forceRoot(filePath);
    
    const pt = new PassThrough();
    conn.client.download(pt, filePath, options.start);
    return pt;
  }
}

您可以找到有关 Node.js 流 here 和 here 的更多信息。

UPD：使用示例：

// assume res is an [express or similar] response object.
const s = store.readStream(file, { start, end });
s.pipe(res);

【讨论】：

我的方法需要返回一个ReadStream，因为我需要访问.on('open', () => stream.pipe(res)) 这是否允许？
@GetOffMyLawn 是的，因为PassThrough 扩展了Transform，它扩展了Duplex，它“扩展”了Readable 和Writable
Type 'PassThrough' is missing the following properties from type 'ReadStream': close, bytesRead, path
@GetOffMyLawn 你使用 TypeScript 吗？
是的，我正在使用打字稿

【解决方案2】：

管道的工作方式与您的想法相反。根据Node.js's documentation，pipe() 是Readable 的方法，它接受Writable 作为其目的地。你试图做的是将Writable 传递给Readable，但实际上它是一个Readable 可以传递给Writeable，而不是相反。

尝试将PassThrough 传递给download() 并返回相同的PassThrough？

【讨论】：