feat(stream): Add pump and prime
MODIFY
packages/stream-types-test/validation.ts
packages/stream-types-test/validation.ts
/* eslint-disable no-unused-vars, no-underscore-dangle, no-empty */
/// <reference types="ses"/>
-import { makeQueue, makeStream, makePipe, mapReader, mapWriter } from '@endo/stream';
+import { makeQueue, makeStream, makePipe, pump, prime, mapReader, mapWriter } from '@endo/stream';
// eslint-disable-next-line
import type { Stream, Reader, Writer } from '@endo/stream';
@@ -49,6 +49,39 @@ async () => {
}
};
+async () => {
+ const [r1, writer] = makePipe<string, number>();
+ const [reader, w2] = makePipe<string, number>();
+ await pump(writer, reader, 1);
+};
+
+async () => {
+ const s: Stream<string> =
+ prime(async function *generator() {
+ yield 'A';
+ return undefined;
+ }());
+ s.return(undefined);
+};
+
+async () => {
+ const s: Stream<string, number> =
+ prime(async function *generator() {
+ const n: number = yield 'A';
+ return undefined;
+ }(), 1);
+ s.return(undefined);
+};
+
+async () => {
+ const s: Stream<string, number, bigint, bigint> =
+ prime(async function *generator() {
+ const n: number = yield 'A';
+ return 1n;
+ }(), 1);
+ s.return(2n);
+};
+
async () => {
const [reader, writer] = makePipe<number, string, boolean, Array<number>>();
const a: IteratorResult<number, boolean> = await reader.next('A');

MODIFY
packages/stream/README.md
packages/stream/README.md
@@ -20,7 +20,7 @@ Awaiting the returned promise slows the writer to match the pace of the reader.
## Reading
-To read from a string, await the value returned by the next method.
+To read from a stream, await the value returned by the next method.
```js
for await (const value of reader) {
@@ -69,6 +69,60 @@ which promises settle.
A stream is consequently a pair of queues that transport iteration results,
one to send messages forward and another to receive acknowledgements.
+## Pump
+
+The `pump` function pumps iterations from a reader to a writer.
+The pump must be primed with the first acknowledgement to send to the reader,
+typically `undefined`, as in `reader.next(undefined)`.
+This makes the parity of a pump "odd", because the reader needs a free
+acknowledgement to start.
+This is in contrast to a pipe, which has "even" parity, because the reader and
+writer can both proceed initially.
+
+So, for example, we can implement `cat` in Node.js by pumping stdin to stdout.
+
+```js
+import { makeNodeWriter, makeNodeReader } from '@endo/stream-node';
+
+const writer = makeNodeWriter(process.stdout);
+const reader = makeNodeReader(process.stdin);
+await pump(writer, reader);
+```
+
+## Prime
+
+Async generator functions are very useful for making reader adapters.
+
+```js
+async function *double(reader) {
+ for await (const value of reader) {
+ yield value * 2;
+ }
+ return undefined;
+}
+```
+
+However, async generator functions can also serve as writers, because `yield`
+evaluates to the argument passed to `next`.
+However, generator writers have odd parity, meaning the first value sent to a
+generator function has nowhere to go and gets discarded as the program counter
+proceeds from the beginning of the function to the first `yield`, `return`, or
+`throw`.
+
+The `prime` function compensates for this by sending a primer to the generator
+once.
+
+```js
+async function *logGenerator() {
+ for (;;) {
+ console.log(yield);
+ }
+}
+
+const writer = prime(logGenerator());
+await writer.next('First message is not discarded');
+```
+
## Hardening
This library depends on Hardened JavaScript.

MODIFY
packages/stream/index.d.ts
packages/stream/index.d.ts
+/* eslint-disable no-unused-vars, import/no-unresolved */
+import { AsyncQueue, Stream } from './types.js';
+
export * from './types.js';
-import { AsyncQueue, Stream, Reader, Writer } from './types.js';
-export function makeQueue<TValue>(): AsyncQueue<TValue>;
+export declare function makeQueue<TValue>(): AsyncQueue<TValue>;
-export function makeStream<
+export declare function makeStream<
TRead,
TWrite = undefined,
TReadReturn = undefined,
- TWriteReturn = undefined,
+ TWriteReturn = undefined
>(
acks: AsyncQueue<IteratorResult<TRead, TReadReturn>>,
data: AsyncQueue<IteratorResult<TWrite, TWriteReturn>>,
): Stream<TRead, TWrite, TReadReturn, TWriteReturn>;
-export function makePipe<
+export declare function makePipe<
TRead,
TWrite = undefined,
TReadReturn = undefined,
- TWriteReturn = undefined,
+ TWriteReturn = undefined
>(): [
Stream<TRead, TWrite, TReadReturn, TWriteReturn>,
Stream<TWrite, TRead, TWriteReturn, TReadReturn>,
];
-export function mapReader<
+export declare function pump<
+ TRead,
+ TWrite = unknown,
+ TReadReturn = unknown,
+ TWriteReturn = unknown
+>(
+ writer: Stream<TWrite, TRead, TWriteReturn, TReadReturn>,
+ reader: Stream<TRead, TWrite, TReadReturn, TWriteReturn>,
+ primer?: TWrite,
+): Promise<void>;
+
+export declare function prime<TRead>(
+ writer: AsyncGenerator<TRead, undefined, undefined>,
+): // primer is implicitly undefined for this overload.
+Stream<TRead, undefined, undefined, undefined>;
+// ESLint hasn't heard about overloads.
+// eslint-disable-next-line no-redeclare
+export declare function prime<TRead, TWrite = unknown, TReturn = unknown>(
+ writer: AsyncGenerator<TRead, TReturn, TWrite>,
+ primer: TWrite,
+): Stream<TRead, TWrite, TReturn, TReturn>;
+
+export declare function mapReader<
TReadIn,
TReadOut = TReadIn,
TWrite = undefined,
TReadReturn = undefined,
- TWriteReturn = undefined,
+ TWriteReturn = undefined
>(
reader: Stream<TReadIn, TWrite, TReadReturn, TWriteReturn>,
transform: (value: TReadIn) => TReadOut,
): Stream<TReadOut, TWrite, TReadReturn, TWriteReturn>;
-export function mapWriter<
+export declare function mapWriter<
TWriteIn,
TWriteOut = TWriteIn,
TRead = undefined,
TReadReturn = undefined,
- TWriteReturn = undefined,
+ TWriteReturn = undefined
>(
writer: Stream<TRead, TWriteOut, TReadReturn, TWriteReturn>,
transform: (value: TWriteIn) => TWriteOut,

MODIFY
packages/stream/index.js
packages/stream/index.js
@@ -115,6 +115,99 @@ export const makePipe = () => {
};
harden(makePipe);
+/**
+ * @template TRead
+ * @template TWrite
+ * @template TReadReturn
+ * @template TWriteReturn
+ * @param {import('./types.js').Stream<TWrite, TRead, TWriteReturn, TReadReturn>} writer
+ * @param {import('./types.js').Stream<TRead, TWrite, TReadReturn, TWriteReturn>} reader
+ * @param {TWrite} primer
+ */
+export const pump = async (writer, reader, primer) => {
+ /** @param {Promise<IteratorResult<TRead, TReadReturn>>} promise */
+ const tick = promise =>
+ promise.then(
+ result => {
+ if (result.done) {
+ return writer.return(result.value);
+ } else {
+ // Behold: mutual recursion.
+ // eslint-disable-next-line no-use-before-define
+ return tock(writer.next(result.value));
+ }
+ },
+ (/** @type {Error} */ error) => {
+ return writer.throw(error);
+ },
+ );
+ /** @param {Promise<IteratorResult<TWrite, TWriteReturn>>} promise */
+ const tock = promise =>
+ promise.then(
+ result => {
+ if (result.done) {
+ return reader.return(result.value);
+ } else {
+ return tick(reader.next(result.value));
+ }
+ },
+ (/** @type {Error} */ error) => {
+ return reader.throw(error);
+ },
+ );
+ await tick(reader.next(primer));
+ return undefined;
+};
+harden(pump);
+
+/**
+ * @template TRead
+ * @template TWrite
+ * @template TReturn
+ * @param {AsyncGenerator<TRead, TReturn, TWrite>} generator
+ * @param {TWrite} primer
+ */
+export const prime = (generator, primer) => {
+ // We capture the first returned promise.
+ const first = generator.next(primer);
+ /** @type {IteratorResult<TRead, TReturn>=} */
+ let result;
+ const primed = harden({
+ /** @param {TWrite} value */
+ async next(value) {
+ if (result === undefined) {
+ result = await first;
+ if (result.done) {
+ return result;
+ }
+ }
+ return generator.next(value);
+ },
+ /** @param {TReturn} value */
+ async return(value) {
+ if (result === undefined) {
+ result = await first;
+ if (result.done) {
+ return result;
+ }
+ }
+ return generator.return(value);
+ },
+ /** @param {Error} error */
+ async throw(error) {
+ if (result === undefined) {
+ result = await first;
+ if (result.done) {
+ throw error;
+ }
+ }
+ return generator.throw(error);
+ },
+ });
+ return primed;
+};
+harden(prime);
+
/**
* @template TIn
* @template TOut

MODIFY
packages/stream/package.json
packages/stream/package.json
"eslintConfig": {
"extends": [
"@endo"
+ ],
+ "overrides": [
+ {
+ "files": [
+ "**/*.{js,ts}"
+ ]
+ }
]
},
"prettier": {

ADD
packages/stream/test/test-prime.js
+// @ts-check
+/* eslint-disable require-yield, no-empty-function */
+
+import 'ses';
+import './lockdown.js';
+
+import rawTest from 'ava';
+import { wrapTest } from '@endo/ses-ava';
+import { prime } from '../index.js';
+
+const test = wrapTest(rawTest);
+
+test('prime single next', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ async function* single() {
+ t.is(yield, 0);
+ }
+
+ const iterator = prime(single());
+ const { done, value } = await iterator.next(0);
+ t.is(done, true);
+ t.is(value, undefined);
+});
+
+test('prime single return', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ async function* empty() {
+ return 'Z';
+ }
+
+ const iterator = prime(empty());
+ const { done, value } = await iterator.return(0);
+ t.is(done, true);
+ t.is(value, 'Z');
+});
+
+test('prime empty throw in', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ async function* empty() {
+ return 'Z';
+ }
+
+ const iterator = prime(empty());
+ try {
+ await iterator.throw(new Error('Abort'));
+ t.fail();
+ } catch (error) {
+ t.is(error.message, 'Abort');
+ }
+});
+
+test('prime single throw in', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ async function* single() {
+ try {
+ t.is(yield, 0);
+ } catch (error) {
+ t.is(error.message, 'Abort');
+ return 'Z';
+ }
+ return 'A';
+ }
+
+ const iterator = prime(single());
+ const { done, value } = await iterator.throw(new Error('Abort'));
+ t.is(done, true);
+ t.is(value, 'Z');
+});
+
+test('prime single throw', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ async function* empty() {
+ throw new Error('Abort');
+ }
+
+ const iterator = prime(empty());
+ try {
+ await iterator.next(0);
+ t.fail('reached beyond end of generator');
+ } catch (error) {
+ t.is(error.message, 'Abort');
+ }
+});
+
+test('prime empty case', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ async function* empty() {}
+
+ const iterator = prime(empty());
+ const { done, value } = await iterator.next();
+ t.is(done, true);
+ t.is(value, undefined);
+});
+
+test('prime throw case', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ async function* temperamental() {
+ throw new Error('Abort');
+ }
+
+ const iterator = prime(temperamental());
+ try {
+ await iterator.next();
+ t.fail();
+ } catch (error) {
+ t.is(error.message, 'Abort');
+ }
+});

ADD
packages/stream/test/test-pump.js
+// @ts-check
+/* global process */
+
+import 'ses';
+import './lockdown.js';
+
+import rawTest from 'ava';
+import { wrapTest } from '@endo/ses-ava';
+import { pump, prime } from '../index.js';
+
+const test = wrapTest(rawTest);
+
+test('pump happy path', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ t.plan(2);
+
+ async function* source() {
+ for (let i = 0; i < 4; i += 1) {
+ yield i;
+ }
+ yield* [4, 5, 6];
+ return '.';
+ }
+
+ async function* target() {
+ const collected = [];
+ try {
+ for (;;) {
+ collected.push(yield);
+ }
+ } catch (error) {
+ t.fail(error);
+ } finally {
+ t.assert(true);
+ t.deepEqual(collected, [0, 1, 2, 3, 4, 5, 6]);
+ }
+ }
+
+ await pump(prime(target()), source());
+});
+
+test('pump target closes early', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ t.plan(7);
+
+ async function* source() {
+ try {
+ for (let i = 0; i < 10; i += 1) {
+ yield i;
+ }
+ return 'X';
+ } finally {
+ t.assert(true);
+ }
+ }
+
+ async function* target() {
+ try {
+ for (let i = 0; i < 5; i += 1) {
+ t.is(yield, i);
+ }
+ return 'Y';
+ } finally {
+ t.assert(true);
+ }
+ }
+
+ await pump(prime(target()), source());
+});
+
+test('pump with error', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ t.plan(3);
+
+ async function* source() {
+ for (let i = 0; i < 4; i += 1) {
+ yield i;
+ }
+ yield* [4, 5, 6];
+ throw new Error('Abort');
+ }
+
+ const collected = [];
+
+ async function* target() {
+ try {
+ for (;;) {
+ collected.push(yield);
+ }
+ } catch (error) {
+ t.is(error.message, 'Abort');
+ t.deepEqual(collected, [0, 1, 2, 3, 4, 5, 6]);
+ } finally {
+ t.assert(true);
+ }
+ }
+
+ try {
+ await pump(prime(target()), source());
+ } catch (error) {
+ // In Node.js 12, the error thrown on the last line of the `source`
+ // generator does not propagate to the promise returned by `pump`.
+ // In Node.js 14, the error does propagate.
+ // We do not assert here so that the plan number is valid in either
+ // version.
+ const major = +process.version.split('.')[0].slice(1);
+ if (major >= 14) {
+ t.fail();
+ }
+ }
+});
+
+test('pump iterator protocol happy', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ t.plan(22);
+
+ let i = 0;
+ const source = {
+ /** @param {undefined} value */
+ async next(value) {
+ t.is(value, undefined);
+ if (i < 10) {
+ t.log(`->${i},`);
+ const next = i;
+ i += 1;
+ return { value: next, done: false };
+ } else {
+ t.log(`->X.`);
+ return { value: 'X', done: true };
+ }
+ },
+ /** @param {undefined} value */
+ async return(value) {
+ t.is(value, undefined);
+ t.log(`->.`);
+ return { value: '?', done: true };
+ },
+ /** @param {Error} error */
+ async throw(error) {
+ t.fail(error.message);
+ t.log(`->!`);
+ return { value: '!', done: true };
+ },
+ [Symbol.asyncIterator]() {
+ return source;
+ },
+ };
+
+ let j = 0;
+ const target = {
+ /** @param {number} value */
+ async next(value) {
+ t.log(`<-${value},`);
+ t.is(value, j);
+ j += 1;
+ return { value: undefined, done: false };
+ },
+ /** @param {string} value */
+ async return(value) {
+ t.log(`<-${value}.`);
+ t.is(value, 'X');
+ return { value: 'Y', done: true };
+ },
+ /** @param {Error} error */
+ async throw(error) {
+ t.log(`<-${error.message}!`);
+ t.fail(error.message);
+ return { value: '!', done: true };
+ },
+ [Symbol.asyncIterator]() {
+ return target;
+ },
+ };
+
+ await pump(target, source);
+});
+
+test('pump iterator protocol source next throws', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ const source = {
+ /** @param {undefined} value */
+ async next(value) {
+ t.log(`->${value},`);
+ throw new Error('Abort');
+ },
+ /** @param {undefined} value */
+ async return(value) {
+ t.log(`->${value}.`);
+ t.fail();
+ return { value: '?', done: true };
+ },
+ /** @param {Error} error */
+ async throw(error) {
+ t.log(`->${error.message}!`);
+ t.fail();
+ return { value: '?', done: true };
+ },
+ [Symbol.asyncIterator]() {
+ return source;
+ },
+ };
+
+ const target = {
+ /** @param {number} value */
+ async next(value) {
+ t.fail();
+ t.log(`<-${value},`);
+ return { value: undefined, done: false };
+ },
+ /** @param {string} value */
+ async return(value) {
+ t.fail();
+ t.log(`<-${value}.`);
+ return { value: undefined, done: true };
+ },
+ /** @param {Error} error */
+ async throw(error) {
+ t.log(`<-${error.message}!`);
+ t.is(error.message, 'Abort');
+ return { value: 'Ack Abort', done: true };
+ },
+ [Symbol.asyncIterator]() {
+ return target;
+ },
+ };
+
+ await pump(target, source);
+});
+
+test('pump iterator protocol target next throws', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ const source = {
+ /** @param {undefined} value */
+ async next(value) {
+ t.is(value, 'A');
+ t.log(`->${value},`);
+ return { value: 0, done: false };
+ },
+ /** @param {undefined} value */
+ async return(value) {
+ t.log(`->${value}.`);
+ t.fail('source return reached');
+ return { value: '?', done: true };
+ },
+ /** @param {Error} error */
+ async throw(error) {
+ t.log(`->${error.message}!`);
+ t.is(error.message, 'Abort');
+ return { value: 'Ack Abort', done: true };
+ },
+ [Symbol.asyncIterator]() {
+ return source;
+ },
+ };
+
+ const target = {
+ /** @param {number} value */
+ async next(value) {
+ t.log(`<-${value},`);
+ t.is(value, 0);
+ throw new Error('Abort');
+ },
+ /** @param {string} value */
+ async return(value) {
+ t.log(`<-${value}.`);
+ t.fail('target return reached');
+ return { value: undefined, done: true };
+ },
+ /** @param {Error} error */
+ async throw(error) {
+ t.log(`<-${error.message}!`);
+ t.fail('target error reached');
+ return { value: '?', done: true };
+ },
+ [Symbol.asyncIterator]() {
+ return target;
+ },
+ };
+
+ await pump(target, source, 'A');
+});
+
+test('pump iterator protocol target return throws', async (/** @type {import('ava').ExecutionContext} */ t) => {
+ const source = {
+ /** @param {undefined} value */
+ async next(value) {
+ t.is(value, 'A');
+ t.log(`->${value},`);
+ return { value: 0, done: true };
+ },
+ /** @param {undefined} value */
+ async return(value) {
+ t.log(`->${value}.`);
+ t.fail('source return reached');
+ return { value: '?', done: true };
+ },
+ /** @param {Error} error */
+ async throw(error) {
+ t.log(`->${error.message}!`);
+ t.fail('source throw reached');
+ return { value: '?', done: true };
+ },
+ [Symbol.asyncIterator]() {
+ return source;
+ },
+ };
+
+ const target = {
+ /** @param {number} value */
+ async next(value) {
+ t.log(`<-${value},`);
+ t.fail('reached target next');
+ return { value: '?', done: true };
+ },
+ /** @param {number} value */
+ async return(value) {
+ t.log(`<-${value}.`);
+ t.is(value, 0);
+ throw new Error('Abort');
+ },
+ /** @param {Error} error */
+ async throw(error) {
+ t.log(`<-${error.message}!`);
+ t.fail('target error reached');
+ return { value: '?', done: true };
+ },
+ [Symbol.asyncIterator]() {
+ return target;
+ },
+ };
+
+ try {
+ await pump(target, source, 'A');
+ t.fail();
+ } catch (error) {
+ t.is(error.message, 'Abort');
+ }
+});

MODIFY
packages/stream/types.d.ts
packages/stream/types.d.ts
+/* eslint-disable no-unused-vars */
+
export interface AsyncQueue<TValue> {
- put(value: TValue | Promise<TValue>): void,
- get(): Promise<TValue>,
+ put(value: TValue | Promise<TValue>): void;
+ get(): Promise<TValue>;
}
// Stream is nearly identical to AsyncGenerator and AsyncGenerator should
@@ -11,13 +13,23 @@ export interface Stream<
TRead,
TWrite = undefined,
TReadReturn = undefined,
- TWriteReturn = undefined,
+ TWriteReturn = undefined
> {
- next(value: TWrite): Promise<IteratorResult<TRead, TReadReturn>>,
- return(value: TWriteReturn): Promise<IteratorResult<TRead, TReadReturn>>,
- throw(error: Error): Promise<IteratorResult<TRead, TReadReturn>>,
- [Symbol.asyncIterator](): Stream<TRead, TWrite, TReadReturn, TWriteReturn>,
+ next(value: TWrite): Promise<IteratorResult<TRead, TReadReturn>>;
+ return(value: TWriteReturn): Promise<IteratorResult<TRead, TReadReturn>>;
+ throw(error: Error): Promise<IteratorResult<TRead, TReadReturn>>;
+ [Symbol.asyncIterator](): Stream<TRead, TWrite, TReadReturn, TWriteReturn>;
}
-export type Reader<TRead, TReadReturn = undefined> = Stream<TRead, undefined, TReadReturn, undefined>;
-export type Writer<TWrite, TWriteReturn = undefined> = Stream<undefined, TWrite, undefined, TWriteReturn>;
+export type Reader<TRead, TReadReturn = undefined> = Stream<
+ TRead,
+ undefined,
+ TReadReturn,
+ undefined
+>;
+export type Writer<TWrite, TWriteReturn = undefined> = Stream<
+ undefined,
+ TWrite,
+ undefined,
+ TWriteReturn
+>;

feat(stream): Add pump and prime