Skip to content

Commit

Permalink
stream: use .chunk when calling adapters's writev
Browse files Browse the repository at this point in the history
Fix: #42157

PR-URL: #42161
Fixes: #42157
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
  • Loading branch information
meixg authored and danielleadams committed Apr 24, 2022
1 parent cd914a5 commit 352cbd0
Show file tree
Hide file tree
Showing 3 changed files with 406 additions and 4 deletions.
10 changes: 6 additions & 4 deletions lib/internal/webstreams/adapters.js
Original file line number Diff line number Diff line change
Expand Up @@ -228,8 +228,9 @@ function newStreamWritableFromWritableStream(writableStream, options = {}) {

writev(chunks, callback) {
function done(error) {
error = error.filter((e) => e);
try {
callback(error);
callback(error.length === 0 ? undefined : error);
} catch (error) {
// In a next tick because this is happening within
// a promise context, and if there are any errors
Expand All @@ -247,7 +248,7 @@ function newStreamWritableFromWritableStream(writableStream, options = {}) {
PromiseAll(
ArrayPrototypeMap(
chunks,
(chunk) => writer.write(chunk))),
(data) => writer.write(data.chunk))),
done,
done);
},
Expand Down Expand Up @@ -633,8 +634,9 @@ function newStreamDuplexFromReadableWritablePair(pair = {}, options = {}) {

writev(chunks, callback) {
function done(error) {
error = error.filter((e) => e);
try {
callback(error);
callback(error.length === 0 ? undefined : error);
} catch (error) {
// In a next tick because this is happening within
// a promise context, and if there are any errors
Expand All @@ -652,7 +654,7 @@ function newStreamDuplexFromReadableWritablePair(pair = {}, options = {}) {
PromiseAll(
ArrayPrototypeMap(
chunks,
(chunk) => writer.write(chunk))),
(data) => writer.write(data.chunk))),
done,
done);
},
Expand Down
166 changes: 166 additions & 0 deletions test/parallel/test-whatwg-webstreams-adapters-to-streamduplex.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,166 @@
// Flags: --no-warnings --expose-internals
'use strict';

const common = require('../common');

const assert = require('assert');

const {
TransformStream,
} = require('stream/web');

const {
newStreamDuplexFromReadableWritablePair,
} = require('internal/webstreams/adapters');

const {
finished,
pipeline,
Readable,
Writable,
} = require('stream');

const {
kState,
} = require('internal/webstreams/util');

{
const transform = new TransformStream();
const duplex = newStreamDuplexFromReadableWritablePair(transform);

assert(transform.readable.locked);
assert(transform.writable.locked);

duplex.destroy();

duplex.on('close', common.mustCall(() => {
assert.strictEqual(transform.readable[kState].state, 'closed');
assert.strictEqual(transform.writable[kState].state, 'errored');
}));
}

{
const error = new Error('boom');
const transform = new TransformStream();
const duplex = newStreamDuplexFromReadableWritablePair(transform);

assert(transform.readable.locked);
assert(transform.writable.locked);

duplex.destroy(error);
duplex.on('error', common.mustCall((reason) => {
assert.strictEqual(reason, error);
}));

duplex.on('close', common.mustCall(() => {
assert.strictEqual(transform.readable[kState].state, 'closed');
assert.strictEqual(transform.writable[kState].state, 'errored');
assert.strictEqual(transform.writable[kState].storedError, error);
}));
}

{
const transform = new TransformStream();
const duplex = new newStreamDuplexFromReadableWritablePair(transform);

duplex.end();
duplex.resume();

duplex.on('close', common.mustCall(() => {
assert.strictEqual(transform.readable[kState].state, 'closed');
assert.strictEqual(transform.writable[kState].state, 'closed');
}));
}

{
const ec = new TextEncoder();
const dc = new TextDecoder();
const transform = new TransformStream({
transform(chunk, controller) {
const text = dc.decode(chunk);
controller.enqueue(ec.encode(text.toUpperCase()));
}
});
const duplex = new newStreamDuplexFromReadableWritablePair(transform, {
encoding: 'utf8',
});

duplex.end('hello');
duplex.on('data', common.mustCall((chunk) => {
assert.strictEqual(chunk, 'HELLO');
}));
duplex.on('end', common.mustCall());

duplex.on('close', common.mustCall(() => {
assert.strictEqual(transform.readable[kState].state, 'closed');
assert.strictEqual(transform.writable[kState].state, 'closed');
}));
}

{
const ec = new TextEncoder();
const dc = new TextDecoder();
const transform = new TransformStream({
transform: common.mustCall((chunk, controller) => {
const text = dc.decode(chunk);
controller.enqueue(ec.encode(text.toUpperCase()));
})
});
const duplex = new newStreamDuplexFromReadableWritablePair(transform, {
encoding: 'utf8',
});

finished(duplex, common.mustCall());

duplex.end('hello');
duplex.resume();
}

{
const ec = new TextEncoder();
const dc = new TextDecoder();
const transform = new TransformStream({
transform: common.mustCall((chunk, controller) => {
const text = dc.decode(chunk);
controller.enqueue(ec.encode(text.toUpperCase()));
})
});
const duplex = new newStreamDuplexFromReadableWritablePair(transform, {
encoding: 'utf8',
});

const readable = new Readable({
read() {
readable.push(Buffer.from('hello'));
readable.push(null);
}
});

const writable = new Writable({
write: common.mustCall((chunk, encoding, callback) => {
assert.strictEqual(dc.decode(chunk), 'HELLO');
assert.strictEqual(encoding, 'buffer');
callback();
})
});

finished(duplex, common.mustCall());
pipeline(readable, duplex, writable, common.mustCall());
}

{
const transform = new TransformStream();
const duplex = newStreamDuplexFromReadableWritablePair(transform);
duplex.setEncoding('utf-8');
duplex.on('data', common.mustCall((data) => {
assert.strictEqual(data, 'hello');
}, 5));

duplex.write(Buffer.from('hello'));
duplex.write(Buffer.from('hello'));
duplex.write(Buffer.from('hello'));
duplex.write(Buffer.from('hello'));
duplex.write(Buffer.from('hello'));

duplex.end();
}
Loading

0 comments on commit 352cbd0

Please sign in to comment.