node.js - Nodejs流暂停(unpipe)和恢复(pipe)中间管道

Question

我需要将可读流“暂停”一定秒数，然后再次恢复。可读流被管道传输到转换流，所以我不能使用常规的 pause 和 resume 方法，我不得不使用 unpipe 和 管道。在转换流中，我能够检测到 pipe 事件，然后在可读流上执行 unpipe，然后在几秒钟后执行 pipe 再次恢复它(我希望)。

代码如下:

主要.ts

import {Transform, Readable} from 'stream';

const alphaTransform = new class extends Transform {
    constructor() {
        super({
            objectMode: true,
            transform: (chunk: string | Buffer, encoding: string, callback: Function) => {
                let transformed: IterableIterator<string>;
                if (Buffer.isBuffer(chunk)) {
                    transformed = function* () {
                        for (const val of chunk) {
                            yield String.fromCharCode(val);
                        }
                    }();
                } else {
                    transformed = chunk[Symbol.iterator]();
                }
                callback(null,
                    Array.from(transformed).map(s => s.toUpperCase()).join(''));
            }
        });
    }
}

const spyingAlphaTransformStream =  new class extends Transform {
    private oncePaused = false;

    constructor() {
        super({
            transform: (chunk: string | Buffer, encoding: string, callback: Function) => {
                console.log('Before transform:');
                if (Buffer.isBuffer(chunk)) {
                    console.log(chunk.toString('utf-8'));
                    alphaTransform.write(chunk);
                } else {
                    console.log(chunk);
                    alphaTransform.write(chunk, encoding);
                }
                callback(null, alphaTransform.read());
            }
        });

        this.on('pipe', (src: Readable) => {
            if (!this.oncePaused) {
                src.unpipe(this); // Here I unpipe the readable stream
                console.log(`Data event listeners count: ${src.listeners('data').length}`);
                console.log(`Readable state of reader: ${src.readable}`);
                console.log("We paused the reader!!");
                setTimeout(() => {
                    this.oncePaused = true;
                    src.pipe(this); // Here I resume it...hopefully?
                    src.resume();
                    console.log("We unpaused the reader!!");
                    console.log(`Data event listeners count: ${src.listeners('data').length}`);
                    console.log(`Readable state of reader: ${src.readable}`);
                }, 1000);
            }
        });

        this.on('data', (transformed) => {
            console.log('After transform:\n', transformed);
        });
    }
}

const reader = new class extends Readable {
    constructor(private content?: string | Buffer) {
        super({
            read: (size?: number) => {
                if (!this.content) {
                    this.push(null);
                } else {
                    this.push(this.content.slice(0, size));
                    this.content = this.content.slice(size);
                }
            }
        });
    }
} (new Buffer('The quick brown fox jumps over the lazy dog.\n'));

reader.pipe(spyingAlphaTransformStream)
    .pipe(process.stdout);

问题出在中间流 spyingAlphaTransformStream 上。这是监听管道事件然后暂停并在 1 秒 后恢复可读流的那个。问题是，在它取消可读流的管道传输，然后再次从中传输后，没有任何内容写入标准输出，这意味着 spyingAlphaTransformStream 的 transform 方法是从未被调用，这意味着流中的某些东西被破坏了。

我希望输出看起来像这样:

Data event listeners count: 0
Readable state of reader: true
We paused the reader!!
We unpaused the reader!!
Data event listeners count: 1
Readable state of reader: true
Before transform:
The quick brown fox jumps over the lazy dog.

After transform:
 THE QUICK BROWN FOX JUMPS OVER THE LAZY DOG.

THE QUICK BROWN FOX JUMPS OVER THE LAZY DOG.

但它实际上看起来像:

Data event listeners count: 0
Readable state of reader: true
We paused the reader!!
We unpaused the reader!!
Data event listeners count: 1
Readable state of reader: true

基本上， 没有通过管道传输任何内容，可读性是我可以从中得出的结论。

我该如何解决这个问题？

package.json

{
  "name": "hello-stream",
  "version": "1.0.0",
  "main": "main.ts",
  "scripts": {
    "start": "npm run build:live",
    "build:live": "nodemon"
  },
  "keywords": [
    "typescript",
    "nodejs",
    "ts-node",
    "cli",
    "node",
    "hello"
  ],
  "license": "WTFPL",
  "devDependencies": {
    "@types/node": "^7.0.21",
    "nodemon": "^1.11.0",
    "ts-node": "^3.0.4",
    "typescript": "^2.3.2"
  },
  "dependencies": {}
}

nodemon.json

{
    "ignore": ["node_modules"],
    "delay": "2000ms",
    "execMap": {
        "ts": "ts-node"
    },
    "runOnChangeOnly": false,
    "verbose": true
}

tsconfig.json

{
  "compilerOptions": {
    "target": "es2015",
    "module": "commonjs",
    "typeRoots": ["node_modules/@types"],
    "lib": ["es6", "dom"],

    "strict": true,
    "noUnusedLocals": true,
    "types": ["node"]
  }
}

Answer 1

解决方案比我预期的要简单得多。我必须做的是找到一种方法来推迟在 transform 方法中完成的任何回调，并等到流“准备好”后再调用初始回调。

基本上，在 spyingAlphaTransformStream 构造函数中，我有一个 bool 值来检查流是否准备就绪，如果没有，我在类中存储一个回调，它将执行第一个回调 I在 transform 方法中接收。 现在，由于第一个回调未执行，流不会收到进一步的调用，即只有一个挂起的回调需要担心；所以它现在只是一个等待游戏，直到流指示它已准备就绪(这是通过一个简单的 setTimeout 完成的)。

当流“就绪”时，我将就绪 bool 值设置为 true，然后我调用挂起的回调(如果已设置)，此时，流将继续贯穿整个流。

我有一个更长的例子来展示它是如何工作的:

import {Transform, Readable} from 'stream';

const alphaTransform = new class extends Transform {
    constructor() {
        super({
            objectMode: true,
            transform: (chunk: string | Buffer, encoding: string, callback: Function) => {
                let transformed: IterableIterator<string>;
                if (Buffer.isBuffer(chunk)) {
                    transformed = function* () {
                        for (const val of chunk) {
                            yield String.fromCharCode(val);
                        }
                    }();
                } else {
                    transformed = chunk[Symbol.iterator]();
                }
                callback(null,
                    Array.from(transformed).map(s => s.toUpperCase()).join(''));
            }
        });
    }
}

class LoggingStream extends Transform {
    private pending: () => void;
    private isReady = false;

    constructor(message: string) {
        super({
            objectMode: true,
            transform: (chunk: string | Buffer, encoding: string, callback: Function) => {
                if (!this.isReady) { // ready flag
                    this.pending = () => { // create a pending callback
                        console.log(message);
                        if (Buffer.isBuffer(chunk)) {
                            console.log(`[${new Date().toTimeString()}]: ${chunk.toString('utf-8')}`);
                        } else {
                            console.log(`[${new Date().toTimeString()}]: ${chunk}`);
                        }
                        callback(null, chunk);
                    }
                } else {
                    console.log(message);
                    if (Buffer.isBuffer(chunk)) {
                        console.log(`[${new Date().toTimeString()}]: ${chunk.toString('utf-8')}`);
                    } else {
                        console.log(`[${new Date().toTimeString()}]: ${chunk}`);
                    }
                    callback(null, chunk);
                }
            }
        });

        this.on('pipe', this.pauseOnPipe);
    }

    private pauseOnPipe() {
        this.removeListener('pipe', this.pauseOnPipe);
        setTimeout(() => {
            this.isReady = true; // set ready flag to true
            if (this.pending) { // execute pending callbacks (if any) 
                this.pending();
            }
        }, 3000); // wait three seconds
    }
}

const reader = new class extends Readable {
    constructor(private content?: string | Buffer) {
        super({
            read: (size?: number) => {
                if (!this.content) {
                    this.push(null);
                } else {
                    this.push(this.content.slice(0, size));
                    this.content = this.content.slice(size);
                }
            }
        });
    }
} (new Buffer('The quick brown fox jumps over the lazy dog.\n'));

reader.pipe(new LoggingStream("Before transformation:"))
    .pipe(alphaTransform)
    .pipe(new LoggingStream("After transformation:"))
    .pipe(process.stdout);

输出

<Waits about 3 seconds...>

Before transformation:
[11:13:53 GMT-0600 (CST)]: The quick brown fox jumps over the lazy dog.

After transformation:
[11:13:53 GMT-0600 (CST)]: THE QUICK BROWN FOX JUMPS OVER THE LAZY DOG.

THE QUICK BROWN FOX JUMPS OVER THE LAZY DOG.

请注意，由于 JS 是单线程的，因此两个详细流在继续之前等待的时间相同