找出流块是否是最后一个

Question

在转换函数中编写 NodeJS 转换流时，我如何知道该块是最后一个块还是没有任何新块到来。

_transform(chunk: any, encoding: string, callback: Function): void {
    // accumulating chunks here to buffer
    // so that I need to do some processing on the whole buffer
    // and I need to understand when to do that
}

所以我需要知道进入 Stream 的块何时结束，对由所有块组成的缓冲区进行一些处理，然后从流中推送处理后的数据。

Answer 1

使用through2

fs.createReadStream('/tmp/important.dat')
  .pipe(through2(
    (chunk, enc, cb) => cb(null, chunk), // transform is a noop
    function (cb) { // FLUSH FUNCTION
      this.push('tacking on an extra buffer to the end');
      cb();
    }
  ))
  .pipe(fs.createWriteStream('/tmp/wut.txt'));   

Answer 2

例子，

const COUNTER_NULL_SYMBOL = Symbol('COUNTER_NULL_SYMBOL');

const Counter = () => {
  let data = COUNTER_NULL_SYMBOL;
  let counter = 1;
  let first = true;

  const counterStream = new Transform({

    objectMode: true,
    decodeStrings: false,
    highWaterMark: 1,

    transform(chunk, encoding, callback) {
      if (data === COUNTER_NULL_SYMBOL) {
        data = chunk;
        return callback();
      } else {
        this.push({data, counter, last: false, first});
        first = false;

        counter++;
        data = chunk;

        return callback();
      }
    },
  });

  counterStream._flush = function (callback) {
    if (data === COUNTER_NULL_SYMBOL) {
      return callback();
    } else {
      this.push({data, counter, last: true, first});
      return callback();
    }
  };

  return counterStream;
};

Answer 3

在_transform你无法确定是否会有更多数据。

根据您的用例，您可以收听end 事件，也可以使用_flush：

Stream: transform._flush(callback):

自定义转换实现可以实现transform._flush() 方法。这将在没有更多写入数据被使用时调用，但在发出 'end' 事件之前，表示可读流结束。

在transform._flush() 实现中，readable.push() 方法可能会被调用零次或多次，视情况而定。刷新操作完成后必须调用回调函数。