Nodejs流暂停（unpipe）和恢复（pipe）中间管道

Question

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

代码如下：

main.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 second 之后恢复可读流的那个。问题是，在它取消可读流的管道传输，然后再次从中传输后，没有任何内容写入标准输出，这意味着永远不会调用 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 方法中完成的任何回调，并等到流为 "ready" 后再调用初始回调。

基本上，在 spyingAlphaTransformStream 构造函数中，我有一个布尔值来检查流是否准备就绪，如果没有，我在 class 中存储了一个回调，它将执行我在 transform 方法中收到的第一个回调。现在因为第一个回调没有被执行，流不会收到进一步的调用也就是说，只有一个挂起的回调需要担心；所以它现在只是一个等待游戏，直到流指示它已准备就绪（这是通过简单的 setTimeout 完成的）。

当流为 "ready" 时，我将就绪布尔值设置为 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 是单线程的，因此两个详细流在继续之前等待的时间相同

Nodejs流暂停（unpipe）和恢复（pipe）中间管道

Nodejs stream pausing (unpipe) and resume (pipe) intermediate pipe

node.js

typescript

ecmascript-6

node.js-stream

main.ts

package.json

nodemon.json

tsconfig.json

输出