可读流不暂停

Question

当我这样说时，我的可读流不会暂停。我遗漏了什么?

我希望输出在一秒钟的持续时间内每秒钟暂停一次，但它没有。

const { Readable, Writable } = require('stream');
const r = new Readable({
read() {
this.push(Math.random().toString());
}
});
const w = new Writable({
write(data, enc, next) {
console.log(data.toString());
setTimeout(next, 10)
}
});
r.pipe(w);
setInterval(() => {
if(r.isPaused()) {
console.log('>>>>>>>>>>> RESUMING')
r.resume();
} else {
console.log('>>>>>>>>>>> PAUSING')
r.pause();
}
}, 1000);

Answer 1

通过使用pipe/unpipe，我能够模仿暂停/恢复行为。

const { Readable, Writable } = require('stream');
const r = new Readable({
highWaterMark: 1,
read() {
const randomNumber = Math.random();
this.push(randomNumber.toString());
}
});
const w = new Writable({
highWaterMark: 1,
write(randomNumber, enc, next) {
console.log(randomNumber.toString())
setTimeout(next, 1000)
}
});
let isPiped = true;
r.pipe(w);
setInterval(() => {
if(isPiped) {
console.log('>>>>>>>>>>>>>>>>>>>>>>>>>>>>> PAUSE')
r.unpipe(w);
isPiped = false
} else {
console.log('>>>>>>>>>>>>>>>>>>>>>>>>>>>>> RESUME')
r.pipe(w);
isPiped = true;
}
}, 10000);