Typescript：有没有更好的方法来获取类型化的流？

Question

对象流很棒，但它们目前没有类型，这意味着您可以将无意义的流连接在一起。嘘！

目标（ish）

class FooReadable extends Readable<Foo> {
  ...
}
class FooWritable extends Writable<Foo> {
  ...
}
class BarWritable extends Writable<Bar> {
  ...
}

const fooReadable = new FooReadable();
const fooWritable = new FooWritable();
const barWritable = new BarWritable();

fooReadable.pipe(fooWritable); // Okay
fooReadable.pipe(barWritable); // Error!

我已经想出了如何使用 ES6 之前样式的构造函数来完成这项工作，但是 我真的很想编写扩展一些抽象类型类的类（比如更多）。

解决方案实际上应该是引擎盖下的原生流——无需从头开始重新实现所有类。

这是我当前的解决方案，使用 Overwrite 替换我不想要的现有类型的位，并使用工厂方法来稍微捏造类型。

import { Writable, WritableOptions, } from "stream";

export function createWritable<T>(
    opts: TypedWritableOptions<T>): TypedWritable<T> {
  return new Writable(opts) as any;
}

export type TypedWritable<T> = Overwrite<Writable, WritableReplacement<T>>;
export type TypedWritableOptions<T> =
    Overwrite<WritableOptions, WritableOptionsReplacement<T>>


// Given types S and D, returns the equivalent of `S & D`, but if any
// properties are shared between S and D, the D versions completely
// replace those in S.
type Overwrite<S, D> = {
  [P in Exclude<keyof S, keyof D>]: S[P]
} & D;

interface WritableReplacement<T> {
  _write(
      chunk: T,
      encoding: string,
      callback: (error?: Error | null) => void,
      ): void;
  _writev?(
      chunks: Array<{ chunk: T, encoding: string }>,
      callback: (error?: Error | null) => void,
      ): void;
}

interface WritableOptionsReplacement<T> {
  write?(
      chunk: T,
      encoding: string,
      callback: (error?: Error | null) => void,
      ): void;
  writev?(
      chunks: Array<{ chunk: T, encoding: string }>,
      callback: (error?: Error | null) => void,
      ): void;
}

Answer 1

所以，这是更好的方法。

首先，定义（在您的代码库中的某处）一个 JS 文件，该文件只是重新导出适当的流类：

// Readable.js
export { Readable } from 'stream';

然后创建一个随附的.d.ts 文件并编写您想要的任何类型定义。这是我的，大部分是从@types/node/index.d.ts 忠实复制的。

// Readable.d.ts
import { BasicCallback } from './core';

export interface Readable<T> extends ReadStream<T> {}
export class Readable<T> {
  constructor(opts?: ReadableOptions<Readable<T>>);

  _read?(size: number): void;
  _destroy?(error: Error | null, callback: BasicCallback): void;
}

export interface ReadStream<T> {
  readable: boolean;
  readonly readableHighWaterMark: number;
  readonly readableLength: number;
  read(size?: number): T;
  setEncoding(encoding: string): this;
  pause(): this;
  resume(): this;
  isPaused(): boolean;
  unpipe<T extends NodeJS.WritableStream>(destination?: T): this;
  unshift(chunk: T): void;
  wrap(oldStream: NodeJS.ReadableStream): this;
  push(chunk: T | null, encoding?: string): boolean;
  destroy(error?: Error): void;

  /**
   * Event emitter
   * The defined events on documents including:
   * 1. close
   * 2. data
   * 3. end
   * 4. readable
   * 5. error
   */
  addListener(event: 'close', listener: () => void): this;
  addListener(event: 'data', listener: (chunk: T) => void): this;
  addListener(event: 'end', listener: () => void): this;
  addListener(event: 'readable', listener: () => void): this;
  addListener(event: 'error', listener: (err: Error) => void): this;
  addListener(event: string | symbol, listener: (...args: any[]) => void): this;

  emit(event: 'close'): boolean;
  emit(event: 'data', chunk: T): boolean;
  emit(event: 'end'): boolean;
  emit(event: 'readable'): boolean;
  emit(event: 'error', err: Error): boolean;
  emit(event: string | symbol, ...args: any[]): boolean;

  on(event: 'close', listener: () => void): this;
  on(event: 'data', listener: (chunk: T) => void): this;
  on(event: 'end', listener: () => void): this;
  on(event: 'readable', listener: () => void): this;
  on(event: 'error', listener: (err: Error) => void): this;
  on(event: string | symbol, listener: (...args: any[]) => void): this;

  once(event: 'close', listener: () => void): this;
  once(event: 'data', listener: (chunk: T) => void): this;
  once(event: 'end', listener: () => void): this;
  once(event: 'readable', listener: () => void): this;
  once(event: 'error', listener: (err: Error) => void): this;
  once(event: string | symbol, listener: (...args: any[]) => void): this;

  prependListener(event: 'close', listener: () => void): this;
  prependListener(event: 'data', listener: (chunk: T) => void): this;
  prependListener(event: 'end', listener: () => void): this;
  prependListener(event: 'readable', listener: () => void): this;
  prependListener(event: 'error', listener: (err: Error) => void): this;
  prependListener(event: string | symbol, listener: (...args: any[]) => void): this;

  prependOnceListener(event: 'close', listener: () => void): this;
  prependOnceListener(event: 'data', listener: (chunk: T) => void): this;
  prependOnceListener(event: 'end', listener: () => void): this;
  prependOnceListener(event: 'readable', listener: () => void): this;
  prependOnceListener(event: 'error', listener: (err: Error) => void): this;
  prependOnceListener(event: string | symbol, listener: (...args: any[]) => void): this;

  removeListener(event: 'close', listener: () => void): this;
  removeListener(event: 'data', listener: (chunk: T) => void): this;
  removeListener(event: 'end', listener: () => void): this;
  removeListener(event: 'readable', listener: () => void): this;
  removeListener(event: 'error', listener: (err: Error) => void): this;
  removeListener(event: string | symbol, listener: (...args: any[]) => void): this;

  [Symbol.asyncIterator](): AsyncIterableIterator<T>;
}

export interface ReadableOptions<This> {
  highWaterMark?: number;
  encoding?: string;
  objectMode?: boolean;
  read?(this: This, size: number): void;
  destroy?(
      this: This,
      error: Error | null,
      callback: BasicCallback,
      ): void;
}

Answer 2

我通过扩展原始类（可读、可写和转换）以不同的方式面对这个问题，例如：

import { Stream } from "stream";

export class GenericTransform<K, T> extends Stream.Transform {
  private processChunk: (chunk: K, enc: string) => Promise<T>;

  constructor(processChunk: (chunk: K, enc: string) => Promise<T>) {
    super({ objectMode: true });
    this.processChunk = processChunk;
  }

  // eslint-disable-next-line no-underscore-dangle
  public async _transform(chunk: K, enc: string, cb: (error?: Error | null) => void): Promise<void> {
    try {
      this.push(await this.processChunk(chunk, enc));
      cb();
    } catch (err) {
      cb(err);
    }
  }
}

export const createTransform = <K, T>(transform: (chunk: K, enc: string) => Promise<T>): GenericTransform<K, T> => new GenericTransform(transform);

使用方法如下：

  inputStream.pipe<GenericTransform<InputType, OutputType>>(
    createTransform<InputType, OutputType>(
      async (chunks: InputType): Promise<OutputType> => {
        await doWhatEver(chunks);
        return chunks;
      }
    )
  ).outputStream

同样的输入流和输出流可以用类似的泛型类来创建以添加类型