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

Question

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

我有以下方法将来自 ftp 服务器的数据写入 block 数组:

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

然后我正在创建一个新的阅读流:

let readable = new ReadStream()

然后我尝试将可写对象通过管道传递给可读对象，但这似乎不起作用:

Argument of type 'ReadStream' is not assignable to parameter of type '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))

Answer 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);