使用 TypeScript 正确扩展 stream.Transform 类

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

好的，现在它似乎可以工作了，我必须将构造函数选项传递给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 类，但您没有实现正确的方法，您应该实现transform._transform(chunk, encoding, callback)，然后打字稿将自动推断所需的类型。由于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)

应用程序代码不得直接调用此函数。它 应该由子类实现，并由内部调用 仅限可读的类方法。

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

自定义转换实现可以实现 transform._flush() 方法。 当没有更多要写入的数据时，将调用此方法。 已消耗，但在发出“结束”事件之前发出信号结束 可读流。

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

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

transform._transform(chunk, encoding, callback)

应用程序代码不得直接调用此函数。它 应该由子类实现，并由内部调用 仅限可读的类方法。

所有的 Transform 流实现都必须提供一个 _transform() 接受输入并产生输出的方法。 transform._transform() 实现处理正在写入的字节，计算输出， 然后使用 可读.push() 方法。

transform.push() 方法可以被调用零次或多次 从单个输入块生成输出，具体取决于要 作为块的结果输出。

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

回调函数只能在当前块被调用时调用 完全消耗。传递给回调的第一个参数必须是 如果在处理输入时发生错误，则为 Error 对象或 否则为空。如果第二个参数传递给回调，它 将被转发到 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 调用 仅限类方法。

这个可选函数将在流关闭之前被调用， 延迟“完成”事件，直到回调被调用。这很有用 在流结束之前关闭资源或写入缓冲数据。