unpipe() and resume() in _transform()

[utftu]

• Node.js Version: 13.*
• OS: mac
• Scope (install, code, runtime, meta, other?):
• Module (and version) (if relevant):

In case of synchronous pipe removal in _transform, the resume method does not resume as expected

const stream = require('stream')

const fs = require('fs');
const readStream = fs.createReadStream('big.txt')
const writeStream = fs.createWriteStream('result.txt');

const transformStream = new class extends stream.Transform {

 _transform(chunk, encoding, callback) {
 readStream.unpipe()
 readStream.resume()
 }
}

readStream.on('end', () => {
 console.log('never print')
})


readStream
 .pipe(transformStream)
 .pipe(writeStream)

Nevertheless, resume operation during an asynchronous call and removal not from the _transform method

Can someone explain this behavior why unpipe () and resume () do not work in the _transform method?

[addaleax]

/cc @ronag

[ronag]

Minimal test:

'use strict';

const common = require('../common');
const { Readable, Transform } = require('stream')

const r = new Readable({
 read () {
 this.push('asd');
 }
});

r.on('end', common.mustCall());

r
 .pipe(new Transform( {
 transform() {
 r.unpipe()
 r.resume()
 }
 }))
 .resume();

[ronag]

@addaleax: I'm seeing something very strange here. Is there any known scenario where a process.nextTick does not get invoked?

I tried adding some logging into Readable:

function resume(stream, state) {
 if (!state.resumeScheduled) {
 state.resumeScheduled = true;
 console.log('resumeScheduled');
 process.nextTick(resume_, stream, state);
 }
}

function resume_(stream, state) {
 console.log('resume_');
 state.resumeScheduled = false;

 if (!state.reading) {
 stream.read(0);
 }

 stream.emit('resume');
 flow(stream);
 if (state.flowing && !state.reading)
 stream.read(0);
}

However the r.resume() call inside the transform in the test will cause "resumeScheduled" to be printed but it never prints "resume_" after that.

Is this even possible or might I be doing something wrong?

[addaleax]

@ronag The only two things I could think of are a) process.exit() or equivalents being called or b) blocking JS execution that prevents the nextTick queue from running.

For debugging Node.js core, it’s sometimes a good idea to use process._rawDebug() rather than console.log(), which doesn’t use streams under the hood.

That being said, if I just add the console.log() calls like you did, it does print output for me.

[ronag]

For debugging Node.js core, it’s sometimes a good idea to use process._rawDebug() rather than console.log()

Oh, thanks for that! I'll keep digging.