Skip to content

Commit

Permalink
stream: catch and forward error from dest.write
Browse files Browse the repository at this point in the history
PR-URL: #55270
Fixes: #54945
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Robert Nagy <[email protected]>
Reviewed-By: Benjamin Gruenbaum <[email protected]>
Reviewed-By: James M Snell <[email protected]>
  • Loading branch information
jakecastelli authored Dec 19, 2024
1 parent 0675e05 commit fd8de67
Show file tree
Hide file tree
Showing 2 changed files with 82 additions and 4 deletions.
13 changes: 9 additions & 4 deletions lib/internal/streams/readable.js
Original file line number Diff line number Diff line change
Expand Up @@ -1004,10 +1004,15 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
src.on('data', ondata);
function ondata(chunk) {
debug('ondata');
const ret = dest.write(chunk);
debug('dest.write', ret);
if (ret === false) {
pause();
try {
const ret = dest.write(chunk);
debug('dest.write', ret);

if (ret === false) {
pause();
}
} catch (error) {
dest.destroy(error);
}
}

Expand Down
73 changes: 73 additions & 0 deletions test/parallel/test-stream-pipe-objectmode-to-non-objectmode.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
'use strict';

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

// Pipeine objects from object mode to non-object mode should throw an error and
// catch by the consumer
{
const objectReadable = Readable.from([
{ hello: 'hello' },
{ world: 'world' },
]);

const passThrough = new Transform({
transform(chunk, _encoding, cb) {
this.push(chunk);
cb(null);
},
});

passThrough.on('error', common.mustCall());

objectReadable.pipe(passThrough);

assert.rejects(async () => {
// eslint-disable-next-line no-unused-vars
for await (const _ of passThrough);
}, /ERR_INVALID_ARG_TYPE/).then(common.mustCall());
}

// The error should be properly forwarded when the readable stream is in object mode,
// the writable stream is in non-object mode, and the data is string.
{
const stringReadable = Readable.from(['hello', 'world']);

const passThrough = new Transform({
transform(chunk, _encoding, cb) {
this.push(chunk);
throw new Error('something went wrong');
},
});

passThrough.on('error', common.mustCall((err) => {
assert.strictEqual(err.message, 'something went wrong');
}));

stringReadable.pipe(passThrough);
}

// The error should be properly forwarded when the readable stream is in object mode,
// the writable stream is in non-object mode, and the data is buffer.
{
const binaryData = Buffer.from('binary data');

const binaryReadable = new Readable({
read() {
this.push(binaryData);
this.push(null);
}
});

const binaryWritable = new Writable({
write(chunk, _encoding, cb) {
throw new Error('something went wrong');
}
});

binaryWritable.on('error', common.mustCall((err) => {
assert.strictEqual(err.message, 'something went wrong');
}));
binaryReadable.pipe(binaryWritable);
}

0 comments on commit fd8de67

Please sign in to comment.