feat(stream): Add mapReader and mapWriter
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 } from '@endo/stream';
+import { makeQueue, makeStream, makePipe, mapReader, mapWriter } from '@endo/stream';
// eslint-disable-next-line
import type { Stream, Reader, Writer } from '@endo/stream';
@@ -58,3 +58,9 @@ async () => {
const e: IteratorResult<string, Array<number>> = await writer.return(true);
const f: IteratorResult<string, Array<number>> = await writer.throw(new Error('Abort'));
};
+
+async () => {
+ const [reader, writer]: [Stream<number, string>, Stream<string, number>] = makePipe<number, string>();
+ const r: Stream<string> = mapReader<number, string>(reader, (n: number) => String.fromCharCode(n));
+ const w: Stream<unknown, string> = mapWriter<string, number>(writer, (value: string) => value.charCodeAt(0));
+};

MODIFY
packages/stream/README.md
packages/stream/README.md
@@ -28,6 +28,27 @@ for await (const value of reader) {
}
```
+## Map
+
+To map a reader to a reader through a synchronous value transform, use `mapReader`.
+
+```js
+const doubleReader = mapReader(singleReader, n => n * 2);
+```
+
+In this example, any value read from doubleReader will be double what was read
+from singleReader.
+
+To map a writer to a writer through a synchronous value transform, use
+`mapWriter`.
+
+```js
+const singleWriter = mapWriter(doubleWriter, n => n * 2);
+```
+
+In this example, any value written to singleWriter will be writ double to
+doubleWriter.
+
## Pipe
The `makePipe` function returns an entangled pair of streams.

MODIFY
packages/stream/index.d.ts
packages/stream/index.d.ts
@@ -22,3 +22,25 @@ export function makePipe<
Stream<TRead, TWrite, TReadReturn, TWriteReturn>,
Stream<TWrite, TRead, TWriteReturn, TReadReturn>,
];
+
+export function mapReader<
+ TReadIn,
+ TReadOut,
+ TReadReturn = unknown,
+ TWrite = unknown,
+ TWriteReturn = unknown,
+>(
+ reader: Stream<TReadIn, TWrite, TReadReturn, TWriteReturn>,
+ transform: (value: TReadIn) => TReadOut,
+): Stream<TReadOut, TWrite, TReadReturn, TWriteReturn>;
+
+export function mapWriter<
+ TWriteIn,
+ TWriteOut,
+ TWriteReturn = unknown,
+ TRead = unknown,
+ TReadReturn = unknown,
+>(
+ writer: Stream<TRead, TWriteOut, TReadReturn, TWriteReturn>,
+ transform: (value: TWriteIn) => TWriteOut,
+): Stream<TRead, TWriteIn, TReadReturn, TWriteReturn>;

MODIFY
packages/stream/index.js
packages/stream/index.js
@@ -114,3 +114,56 @@ export const makePipe = () => {
return harden([writer, reader]);
};
harden(makePipe);
+
+/**
+ * @template TIn
+ * @template TOut
+ * @param {import('./types.js').Reader<TIn>} reader
+ * @param {(value: TIn) => TOut} transform
+ * @returns {import('./types.js').Reader<TOut>}
+ */
+export const mapReader = (reader, transform) => {
+ async function* transformGenerator() {
+ for await (const value of reader) {
+ yield transform(value);
+ }
+ return undefined;
+ }
+ return harden(transformGenerator());
+};
+harden(mapReader);
+
+/**
+ * @template TIn
+ * @template TOut
+ * @param {import('./types.js').Writer<TOut>} writer
+ * @param {(value: TIn) => TOut} transform
+ * @returns {import('./types.js').Writer<TIn>}
+ */
+export const mapWriter = (writer, transform) => {
+ const transformedWriter = harden({
+ /**
+ * @param {TIn} value
+ */
+ async next(value) {
+ return writer.next(transform(value));
+ },
+ /**
+ * @param {Error} error
+ */
+ async throw(error) {
+ return writer.throw(error);
+ },
+ /**
+ * @param {undefined} value
+ */
+ async return(value) {
+ return writer.return(value);
+ },
+ [Symbol.asyncIterator]() {
+ return transformedWriter;
+ },
+ });
+ return transformedWriter;
+};
+harden(mapWriter);

feat(stream): Add mapReader and mapWriter