node.js - Nodejs-strøm pausing (unpipe) og gjenoppta (pipe) mellomliggende pipe
Jeg trenger å "pause" en lesbar strøm i et bestemt antall sekunder og så gjenoppta den igjen. Den lesbare strømmen sendes til en transformasjonsstrøm, så jeg kan ikke bruke de vanlige pause- og resume-metodene, jeg måtte bruke unpipe og pipe. I transformasjonsstrømmen klarer jeg å oppdage pipe-hendelsen og deretter gjøre unpipe på den lesbare strømmen, og etter noen sekunder gjøre pipe igjen for å gjenoppta den (jeg håper).
Her er koden:
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);
Problemet ligger i mellomstrømmen spyingAlphaTransformStream. Dette er den som lytter etter pipe-hendelsen og deretter pauser og gjenopptar den lesbare strømmen etter 1 sekund. Problemet er at etter at den kobler fra den lesbare strømmen, og deretter kobler til fra den igjen, blir ingenting skrevet til standard utgang, noe som betyr at transform-metoden til spyingAlphaTransformStream aldri blir kalt, noe som betyr at noe er ødelagt i strømmen.
Jeg forventer at resultatet ser omtrent slik ut:
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.
Men det ser faktisk ut som:
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
I utgangspunktet blir ingenting sendt fra det lesbare, er det jeg kan konkludere med ut fra det.
Hvordan kan jeg fikse dette?
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"]
}
}
Svar #1
Løsningen var overraskende enklere enn jeg hadde forventet. Det jeg måtte gjøre, var å finne en måte å utsette eventuelle tilbakeringinger gjort i transform-metoden, og vente til strømmen var «klar» før jeg kalte den opprinnelige tilbakeringen.
I utgangspunktet, i spyingAlphaTransformStream-konstruktøren, hadde jeg en boolsk verdi som sjekket om strømmen var klar, og hvis den ikke var det, lagret jeg en callback i klassen som vil utføre den første callbacken jeg mottok i transform-metoden. Nå siden den første callbacken ikke ble utført, mottar ikke strømmen flere kall dvs. det er bare én ventende callback å bekymre seg for; så det ble nå rett og slett et ventespill til strømmen indikerer at den var klar (dette gjøres med en enkel setTimeout).
Når strømmen er "klar", setter jeg den boolske verdien for klar til sann, deretter kaller jeg den ventende callbacken (hvis satt) og på dette tidspunktet fortsetter flyten gjennom hele strømmen.
Jeg har et lengre eksempel for å vise hvordan dette fungerer:
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);
Utdata
<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.
Merk at siden JS er enkelttrådet, venter begge detaljerte strømmer samme tid før de fortsetter