nodejs流暂停(脱离)和简历(管道)中间管

Question

我需要"暂停"一个可读的流以一定数量的秒数，然后再次恢复。可读的流被管道输送到变换流，因此我无法使用常规的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方法中完成的所有回调，然后等到流到"准备好"之前，请拨打初始回调。

。

基本上，在spyingAlphaTransformStream构造函数中，我有一个布尔值检查流是否准备就绪，如果还没有，我在同类中存储了一个回调，该回调将执行我在transform方法中收到的第一个回调。现在，由于没有执行第一个回调，因此流未接收其他呼叫 即，只有一个待定的回调可以担心；因此，现在只是一个等待游戏，直到流表示已准备就绪(这是用简单的setTimeout完成的(。

当流"准备就绪"时，我将准备就绪的布尔值设置为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是单线螺纹，因此两个详细的流都等待相同的时间，然后继续