找出流块是否是最后一个

Question

在Transform函数中编写NodeJS Transform stream时，我如何知道该块是最后一块还是没有任何新块出现。

js prettyprint-override">_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

在\u转换中无法确定是否会有更多数据。

根据您的用例，您可以收听end事件，或者使用\u flush：

流：转换_刷新（回调）：

自定义Transform实现可以实现转换。_flush（）方法。当没有更多的写入数据要消耗时，但在发出通知可读流结束的'end'事件之前，将调用此选项。

在_flush（）实现中，readable.push（）方法可以被调用0次或更多次。刷新操作完成时必须调用回调函数。