refactor(stream-node): Implement in terms of mapReader
MODIFY
packages/stream-node/reader.js
packages/stream-node/reader.js
// They however iterate Node.js Buffer values and are not hardened, so this
// implementation compensates for both.
+import { mapReader } from '@endo/stream';
+
const { details: X, quote: q } = assert;
/**
@@ -24,43 +26,34 @@ export const makeNodeReader = input => {
)} to a AsyncIterator<Uint8Array>`,
);
- const finalIteration = new Promise((resolve, reject) => {
- input.on('error', reject);
- input.on('close', () => {
- resolve({ done: true, value: undefined });
- });
- });
-
const iterator = input[Symbol.asyncIterator]();
- /** @type {import('@endo/stream').Reader<Uint8Array>} */
- const reader = harden({
+ assert(iterator.return);
+
+ // Adapt the AsyncIterator to the more strict interface of a Stream: must
+ // have return and throw methods.
+ /** @type {import('@endo/stream').Reader<Buffer>} */
+ const reader = {
async next() {
- const result = await iterator.next();
- if (result.done) {
- return result;
- }
- assert(typeof result.value !== 'string');
- const { buffer, byteOffset, length } = result.value;
- return {
- done: false,
- value: new Uint8Array(buffer, byteOffset, length),
- };
+ return iterator.next();
},
async return() {
- input.destroy();
- return finalIteration;
+ assert(iterator.return);
+ return iterator.return();
},
- /**
- * @param {Error} error
- */
async throw(error) {
input.destroy(error);
- return finalIteration;
+ assert(iterator.return);
+ return iterator.return();
},
[Symbol.asyncIterator]() {
return reader;
},
+ };
+
+ /** @type {import('@endo/stream').Reader<Uint8Array>} */
+ return mapReader(reader, buffer => {
+ assert(typeof buffer !== 'string');
+ return new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.length);
});
- return reader;
};
harden(makeNodeReader);

MODIFY
packages/stream-node/test/test-stream-node.js
packages/stream-node/test/test-stream-node.js
@@ -84,82 +84,6 @@ test('stream to and from Node.js reader/writer', async (/** @type {import('ava')
await Promise.all([makeProducer(), makeConsumer(), exited]);
});
-test('stream read return', async (/** @type {import('ava').ExecutionContext} */ t) => {
- const scratch = new Uint8Array(1024 * 128);
- for (let i = 0; i < scratch.byteLength; i += 1) {
- scratch[i] = i % 256;
- }
-
- const child = fork(catPath, {
- // Note: ignoring stderr. Node.js barfs as expected when we close its
- // output stream in this test.
- stdio: ['pipe', 'pipe', 'ignore', 'ipc'],
- });
- const reader = makeNodeReader(child.stdout);
- const writer = makeNodeWriter(child.stdin);
-
- const exited = new Promise((resolve, reject) => {
- child.on('error', reject);
- child.on('exit', resolve);
- });
-
- // Begin forwarding whenever ready:
- child.on('message', () => {
- child.send({});
- });
-
- const makeProducer = async () => {
- await writer.next(scratch);
- await writer.return();
- };
-
- const makeConsumer = async () => {
- // Consumer aborts the stream.
- await reader.return();
- };
-
- await Promise.allSettled([makeProducer(), makeConsumer(), exited]);
- t.pass();
-});
-
-test('stream read throw', async (/** @type {import('ava').ExecutionContext} */ t) => {
- const scratch = new Uint8Array(1024 * 128);
- for (let i = 0; i < scratch.byteLength; i += 1) {
- scratch[i] = i % 256;
- }
-
- const child = fork(catPath, {
- // Note: ignoring stderr. Node.js barfs as expected when we close its
- // output stream in this test.
- stdio: ['pipe', 'pipe', 'ignore', 'ipc'],
- });
- const reader = makeNodeReader(child.stdout);
- const writer = makeNodeWriter(child.stdin);
-
- const exited = new Promise((resolve, reject) => {
- child.on('error', reject);
- child.on('exit', resolve);
- });
-
- // Begin forwarding whenever ready:
- child.on('message', () => {
- child.send({});
- });
-
- const makeProducer = async () => {
- await writer.next(scratch);
- await writer.return();
- };
-
- const makeConsumer = async () => {
- // Consumer aborts the stream.
- await reader.throw(new Error('Abort!'));
- };
-
- await Promise.allSettled([makeProducer(), makeConsumer(), exited]);
- t.pass();
-});
-
test('stream write error (EPIPE due to exit)', async (/** @type {import('ava').ExecutionContext} */ t) => {
const scratch = new Uint8Array(1024 * 128);
for (let i = 0; i < scratch.byteLength; i += 1) {

MODIFY
packages/stream-types-test/validation.ts
packages/stream-types-test/validation.ts
@@ -58,7 +58,7 @@ async () => {
async () => {
const s: Stream<string> =
prime(async function *generator() {
- yield 'A';
+ const u: undefined = yield 'A';
return undefined;
}());
s.return(undefined);

refactor(stream-node): Implement in terms of mapReader