使用 TypeScript 正确扩展 stream.Transform class

Question

我有这个代码：

import * as stream from 'stream';    

export class JSONParser extends stream.Transform {

  lastLineData = '';
  objectMode = true;

  constructor() {
    super();

  }

  transform(chunk, encoding, cb) {

    let data = String(chunk);
    if (this.lastLineData) {
      data = this.lastLineData + data;
    }

    let lines = data.split('\n');
    this.lastLineData = lines.splice(lines.length - 1, 1)[0];

    lines.forEach(l => {
      try {
        // l might be an empty string; ignore if so
        l && this.push(JSON.parse(l));
      }
      catch (err) {
        // noop
      }
    });

    cb();

  }

  flush(cb) {
    if (this.lastLineData) {
      try {
        this.push(JSON.parse(this.lastLineData));
      }
      catch (err) {
        // noop
      }
    }
    this.lastLineData = '';
    cb();
  }


}

问题是 TS 类型无法识别（原型）方法。我是否错误地扩展了转换 class？

这是问题所在：

请注意这是正确的：

import * as stream from 'stream';

//////////////////////////////////////////////////

export interface IParsedObject {
  [index: string]: any
}

export const createParser = function () {

  let lastLineData = '';

  return new stream.Transform({

    objectMode: true,

    transform(chunk: any, encoding: string, cb: Function) {

      let data = String(chunk);
      if (lastLineData) {
        data = lastLineData + data;
      }

      let lines = data.split('\n');
      lastLineData = lines.splice(lines.length - 1, 1)[0];

      lines.forEach(l => {
        try {
          // l might be an empty string; ignore if so
          l && this.push(JSON.parse(l));
        }
        catch (err) {
          // noop
        }
      });

      cb();

    },

    flush(cb: Function) {
      if (lastLineData) {
        try {
          this.push(JSON.parse(lastLineData));
        }
        catch (err) {
          // noop
        }
      }
      lastLineData = '';
      cb();
    }
  });

};

但是上面的 class 似乎不一样。

Answer 1

好的，现在它似乎可以工作了，我不得不将构造函数选项传递给 super()，所以它变成了 super({objectMode:true})、

export class JSONParser extends stream.Transform {

  lastLineData = '';

  constructor() {
    super({objectMode: true});

  }

  _transform(chunk: any, encoding: string, cb: Function) {

    let data = String(chunk);
    if (this.lastLineData) {
      data = this.lastLineData + data;
    }

    let lines = data.split('\n');
    this.lastLineData = lines.splice(lines.length - 1, 1)[0];

    lines.forEach(l => {
      try {
        // l might be an empty string; ignore if so
        l && this.push(JSON.parse(l));
      }
      catch (err) {
        // noop
      }
    });

    cb();

  }

  flush(cb: Function) {
    if (this.lastLineData) {
      try {
        this.push(JSON.parse(this.lastLineData));
      }
      catch (err) {
        // noop
      }
    }
    this.lastLineData = '';
    cb();
  }

  _flush(cb: Function) {
    if (this.lastLineData) {
      try {
        this.push(JSON.parse(this.lastLineData));
      }
      catch (err) {
        // noop
      }
    }
    this.lastLineData = '';
    cb();
  }

}

剩下的唯一问题是我是否应该实施 _flush() 或 flush()，我不确定...我现在实施了两者...

Answer 2

在您的屏幕截图中，您试图扩展 Transform class，但您没有实现正确的方法，您应该实现 transform._transform(chunk, encoding, callback)，然后 typescript 将自动推断所需的类型。由于 transform.transform(chunk, encoding, callback) 在 Transform 类型上不存在，因此打字稿没有可用于推理的类型，因此编译器会生成警告。

在您的代码示例中，您选择使用 "simplified" Transform 构造函数。构造函数采用一组选项，允许您定义必要的转换方法而无需显式扩展 class。这些选项选择非下划线前缀命名约定，但在扩展 class 时它们等同于下划线前缀方法。由于您在此处为方法使用了正确的名称，因此能够推断出类型。

Transform API 调用了三个方法的执行，这些方法概述了 here.

有两种方法可以实现 Transform 流：

扩展 Transform class
使用Transform构造函数选项。

概述了这两种方法here（包括在 ES6 之前的环境中扩展 class 的示例）。

实现 Transform 流时，只需实现一个方法：

transform._transform(chunk, encoding, callback)

另外两个方法是可选的，如果用例需要可以实现：

transform._flush(callback)
writable._final(callback)

我已经在下面概述了记录的 Transform 方法，突出显示了一些可能感兴趣的领域。

`transform._flush(callback)`

This function MUST NOT be called by application code directly. It should be implemented by child classes, and called by the internal Readable class methods only.

In some cases, a transform operation may need to emit an additional bit of data at the end of the stream. For example, a zlib compression stream will store an amount of internal state used to optimally compress the output. When the stream ends, however, that additional data needs to be flushed so that the compressed data will be complete.

Custom Transform implementations may implement the transform._flush() method. This will be called when there is no more written data to be consumed, but before the 'end' event is emitted signaling the end of the Readable stream.

Within the transform._flush() implementation, the readable.push() method may be called zero or more times, as appropriate. The callback function must be called when the flush operation is complete.

The transform._flush() method is prefixed with an underscore because it is internal to the class that defines it, and should never be called directly by user programs.

`transform._transform(chunk, encoding, callback)`

This function MUST NOT be called by application code directly. It should be implemented by child classes, and called by the internal Readable class methods only.

All Transform stream implementations must provide a _transform() method to accept input and produce output. The transform._transform() implementation handles the bytes being written, computes an output, then passes that output off to the readable portion using the readable.push() method.

The transform.push() method may be called zero or more times to generate output from a single input chunk, depending on how much is to be output as a result of the chunk.

It is possible that no output is generated from any given chunk of input data.

The callback function must be called only when the current chunk is completely consumed. The first argument passed to the callback must be an Error object if an error occurred while processing the input or null otherwise. If a second argument is passed to the callback, it will be forwarded on to the readable.push() method. In other words the following are equivalent:

transform.prototype._transform = function(data, encoding, callback) {
  this.push(data);
  callback();
};

transform.prototype._transform = function(data, encoding, callback) {
  callback(null, data);
};

The transform._transform() method is prefixed with an underscore because it is internal to the class that defines it, and should never be called directly by user programs.

transform._transform() is never called in parallel; streams implement a queue mechanism, and to receive the next chunk, callback must be called, either synchronously or asynchronously.

`writable._final(callback)`

The _final() method must not be called directly. It may be implemented by child classes, and if so, will be called by the internal Writable class methods only.

This optional function will be called before the stream closes, delaying the 'finish' event until callback is called. This is useful to close resources or write buffered data before a stream ends.

使用 TypeScript 正确扩展 stream.Transform class

Properly extend stream.Transform class with TypeScript

node.js

typescript

tsc

typescript3.0

`transform._flush(callback)`

`transform._transform(chunk, encoding, callback)`

`writable._final(callback)`