正确扩展stream.Transform类与TypeScript

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类型不能识别（原型）方法。我是否错误地扩展了Transform类？

这是问题所在：

请注意，这是正确的：

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();
    }
  });

};

但上面的类似乎不一样。

Answer 1

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

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

---

Transform API要求实现三种方法，这些方法概述了here。

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

1. 扩展Transform类
2. 使用Transform构造函数选项。

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

---

实现Transform流时，只能实现一种方法：

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

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

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

---

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

transform._flush(callback)

该功能不能直接由应用程序代码调用。它应该由子类实现，并且只能由内部的Readable类方法调用。

在某些情况下，转换操作可能需要在流的末尾发出额外的数据位。例如，zlib压缩流将存储用于最佳压缩输出的内部状态量。但是，当流结束时，需要刷新附加数据，以便完成压缩数据。

自定义转换实现可以实现transform._flush（）方法。当没有更多要写入的数据被消耗时，但在发出'end'事件之前，将发出信号，表示可读流的结束。

在transform._flush（）实现中，可以根据需要将readable.push（）方法调用零次或多次。刷新操作完成后，必须调用回调函数。

transform._flush（）方法以下划线为前缀，因为它是定义它的类的内部，并且永远不应由用户程序直接调用。

transform._transform(chunk, encoding, callback)

该功能不能直接由应用程序代码调用。它应该由子类实现，并且只能由内部的Readable类方法调用。

所有Transform流实现都必须提供_transform（）方法来接受输入并生成输出。 transform._transform（）实现处理正在写入的字节，计算输出，然后使用readable.push（）方法将该输出传递给可读部分。

transform.push（）方法可以被调用零次或多次以从单个输入块生成输出，这取决于作为块的结果将输出多少。

可能没有从任何给定的输入数据块生成输出。

只有在完全消耗当前块时才必须调用回调函数。如果在处理输入时发生错误，则传递给回调的第一个参数必须是Error对象，否则为null。如果将第二个参数传递给回调，它将被转发到readable.push（）方法。换句话说，以下是等效的：

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

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

transform._transform（）方法以下划线为前缀，因为它是定义它的类的内部，并且永远不应由用户程序直接调用。

从不并行调用transform._transform（）;流实现了一个队列机制，并且为了接收下一个块，必须同步或异步调用回调。

writable._final(callback)

不能直接调用_final（）方法。它可以由子类实现，如果是这样，将仅由内部Writable类方法调用。

在流关闭之前将调用此可选函数，延迟'finish'事件，直到调用回调。这对于在流结束之前关闭资源或写入缓冲数据很有用。

Answer 2

好吧所以它现在似乎工作，我不得不将构造函数选项传递给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()，我不确定......我现在实施了两个......