我需要将可读流“暂停”一定秒数，然后再次恢复。可读流被管道传输到转换流，所以我不能使用常规的 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"]
}
}
最佳答案
解决方案比我预期的要简单得多。我必须做的是找到一种方法来推迟在 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 是单线程的，因此两个详细流在继续之前等待的时间相同
关于node.js - Nodejs流暂停(unpipe)和恢复(pipe)中间管道，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/44765765/