Add map, flatMap, split and join streams

src/stream.spec.ts

@@ -1,8 +1,15 @@
 import test from "ava";
 import { expect } from "chai";
 import { Readable } from "stream";
-import { fromArray, collect, concat } from "./stream";
-
+import {
+    fromArray,
+    map,
+    flatMap,
+    split,
+    join,
+    collect,
+    concat,
+} from "./stream";
 test.cb("fromArray() streams array elements in flowing mode", t => {
     t.plan(3);
     const elements = ["a", "b", "c"];
@@ -45,6 +52,269 @@ test.cb("fromArray() ends immediately if there are no array elements", t => {
         .on("end", t.end);
 });
 
+test.cb("map() maps elements synchronously", t => {
+    t.plan(3);
+    const source = new Readable({ objectMode: true });
+    const expectedElements = ["A", "B", "C"];
+    let i = 0;
+    source
+        .pipe(map((element: string) => element.toUpperCase()))
+        .on("data", element => {
+            expect(element).to.equal(expectedElements[i]);
+            t.pass();
+            i++;
+        })
+        .on("error", t.end)
+        .on("end", t.end);
+
+    source.push("a");
+    source.push("b");
+    source.push("c");
+    source.push(null);
+});
+
+test.cb("map() maps elements asynchronously", t => {
+    t.plan(3);
+    const source = new Readable({ objectMode: true });
+    const expectedElements = ["A", "B", "C"];
+    let i = 0;
+    source
+        .pipe(
+            map(async (element: string) => {
+                await Promise.resolve();
+                return element.toUpperCase();
+            }),
+        )
+        .on("data", element => {
+            expect(element).to.equal(expectedElements[i]);
+            t.pass();
+            i++;
+        })
+        .on("error", t.end)
+        .on("end", t.end);
+
+    source.push("a");
+    source.push("b");
+    source.push("c");
+    source.push(null);
+});
+
+test.cb("map() emits errors during synchronous mapping", t => {
+    t.plan(2);
+    const source = new Readable({ objectMode: true });
+    source
+        .pipe(
+            map((element: string) => {
+                if (element !== "a") {
+                    throw new Error("Failed mapping");
+                }
+                return element.toUpperCase();
+            }),
+        )
+        .resume()
+        .on("error", err => {
+            expect(err.message).to.equal("Failed mapping");
+            t.pass();
+        })
+        .on("end", t.end);
+
+    source.push("a");
+    source.push("b");
+    source.push("c");
+    source.push(null);
+});
+
+test.cb("map() emits errors during asynchronous mapping", t => {
+    t.plan(2);
+    const source = new Readable({ objectMode: true });
+    source
+        .pipe(
+            map(async (element: string) => {
+                await Promise.resolve();
+                if (element !== "a") {
+                    throw new Error("Failed mapping");
+                }
+                return element.toUpperCase();
+            }),
+        )
+        .resume()
+        .on("error", err => {
+            expect(err.message).to.equal("Failed mapping");
+            t.pass();
+        })
+        .on("end", t.end);
+
+    source.push("a");
+    source.push("b");
+    source.push("c");
+    source.push(null);
+});
+
+test.cb("flatMap() maps elements synchronously", t => {
+    t.plan(6);
+    const source = new Readable({ objectMode: true });
+    const expectedElements = ["a", "A", "b", "B", "c", "C"];
+    let i = 0;
+    source
+        .pipe(flatMap((element: string) => [element, element.toUpperCase()]))
+        .on("data", (element: string) => {
+            expect(element).to.equal(expectedElements[i]);
+            t.pass();
+            i++;
+        })
+        .on("end", t.end);
+
+    source.push("a");
+    source.push("b");
+    source.push("c");
+    source.push(null);
+});
+
+test.cb("flatMap() maps elements asynchronously", t => {
+    t.plan(6);
+    const source = new Readable({ objectMode: true });
+    const expectedElements = ["a", "A", "b", "B", "c", "C"];
+    let i = 0;
+    source
+        .pipe(
+            flatMap(async (element: string) => {
+                await Promise.resolve();
+                return [element, element.toUpperCase()];
+            }),
+        )
+        .on("data", (element: string) => {
+            expect(element).to.equal(expectedElements[i]);
+            t.pass();
+            i++;
+        })
+        .on("end", t.end);
+
+    source.push("a");
+    source.push("b");
+    source.push("c");
+    source.push(null);
+});
+
+test.cb("flatMap() emits errors during synchronous mapping", t => {
+    t.plan(2);
+    const source = new Readable({ objectMode: true });
+    source
+        .pipe(
+            flatMap((element: string) => {
+                if (element !== "a") {
+                    throw new Error("Failed mapping");
+                }
+                return [element, element.toUpperCase()];
+            }),
+        )
+        .resume()
+        .on("error", err => {
+            expect(err.message).to.equal("Failed mapping");
+            t.pass();
+        })
+        .on("end", t.end);
+
+    source.push("a");
+    source.push("b");
+    source.push("c");
+    source.push(null);
+});
+
+test.cb("flatMap() emits errors during asynchronous mapping", t => {
+    t.plan(2);
+    const source = new Readable({ objectMode: true });
+    source
+        .pipe(
+            flatMap(async (element: string) => {
+                await Promise.resolve();
+                if (element !== "a") {
+                    throw new Error("Failed mapping");
+                }
+                return [element, element.toUpperCase()];
+            }),
+        )
+        .resume()
+        .on("error", err => {
+            expect(err.message).to.equal("Failed mapping");
+            t.pass();
+        })
+        .on("end", t.end);
+
+    source.push("a");
+    source.push("b");
+    source.push("c");
+    source.push(null);
+});
+
+test.cb("split() splits chunks using the default separator (\\n)", t => {
+    t.plan(5);
+    const source = new Readable({ objectMode: true });
+    const expectedParts = ["ab", "c", "d", "ef", ""];
+    let i = 0;
+    source
+        .pipe(split())
+        .on("data", part => {
+            expect(part).to.equal(expectedParts[i]);
+            t.pass();
+            i++;
+        })
+        .on("error", t.end)
+        .on("end", t.end);
+
+    source.push("ab\n");
+    source.push("c");
+    source.push("\n");
+    source.push("d");
+    source.push("\nef\n");
+    source.push(null);
+});
+
+test.cb("split() splits chunks using the specified separator", t => {
+    t.plan(6);
+    const source = new Readable({ objectMode: true });
+    const expectedParts = ["ab", "c", "d", "e", "f", ""];
+    let i = 0;
+    source
+        .pipe(split("|"))
+        .on("data", part => {
+            expect(part).to.equal(expectedParts[i]);
+            t.pass();
+            i++;
+        })
+        .on("error", t.end)
+        .on("end", t.end);
+
+    source.push("ab|");
+    source.push("c|d");
+    source.push("|");
+    source.push("e");
+    source.push("|f|");
+    source.push(null);
+});
+
+test.cb("join() joins chunks using the specified separator", t => {
+    t.plan(9);
+    const source = new Readable({ objectMode: true });
+    const expectedParts = ["ab|", "|", "c|d", "|", "|", "|", "e", "|", "|f|"];
+    let i = 0;
+    source
+        .pipe(join("|"))
+        .on("data", part => {
+            expect(part).to.equal(expectedParts[i]);
+            t.pass();
+            i++;
+        })
+        .on("error", t.end)
+        .on("end", t.end);
+
+    source.push("ab|");
+    source.push("c|d");
+    source.push("|");
+    source.push("e");
+    source.push("|f|");
+    source.push(null);
+});
+
 test.cb(
     "collect() collects streamed elements into an array (object, flowing mode)",
     t => {

src/stream.ts

@@ -20,7 +20,122 @@ export function fromArray(array: any[]): NodeJS.ReadableStream {
 }
 
 /**
- * Return a ReadWrite stream that collects streamed objects or bytes into an array or buffer
+ * Return a ReadWrite stream that maps streamed chunks
+ * @param mapper The mapper function, mapping each (chunk, encoding) to a new chunk (or a promise of such)
+ * @param options
+ * @param options.readableObjectMode Whether this stream should behave as a readable stream of objects
+ * @param options.writableObjectMode Whether this stream should behave as a writable stream of objects
+ */
+export function map<T, R>(
+    mapper: (chunk: T, encoding: string) => R,
+    { readableObjectMode = true, writableObjectMode = true } = {},
+): NodeJS.ReadWriteStream {
+    return new Transform({
+        readableObjectMode,
+        writableObjectMode,
+        async transform(chunk, encoding, callback) {
+            let isPromise = false;
+            try {
+                const mapped = mapper(chunk, encoding);
+                isPromise = mapped instanceof Promise;
+                callback(undefined, await mapped);
+            } catch (err) {
+                if (isPromise) {
+                    // Calling the callback asynchronously with an error wouldn't emit the error, so emit directly
+                    this.emit("error", err);
+                    callback();
+                } else {
+                    callback(err);
+                }
+            }
+        },
+    });
+}
+
+type FlatMapper<T, R> =
+    | ((chunk: T, encoding: string) => R[])
+    | ((chunk: T, encoding: string) => Promise<R[]>);
+/**
+ * Return a ReadWrite stream that flat maps streamed chunks
+ * @param mapper The mapper function, mapping each (chunk, encoding) to an array of new chunks (or a promise of such)
+ * @param options
+ * @param options.readableObjectMode Whether this stream should behave as a readable stream of objects
+ * @param options.writableObjectMode Whether this stream should behave as a writable stream of objects
+ */
+export function flatMap<T, R>(
+    mapper: FlatMapper<T, R>,
+    { readableObjectMode = true, writableObjectMode = true } = {},
+): NodeJS.ReadWriteStream {
+    return new Transform({
+        readableObjectMode,
+        writableObjectMode,
+        async transform(chunk, encoding, callback) {
+            let isPromise = false;
+            try {
+                const mapped = mapper(chunk, encoding);
+                isPromise = mapped instanceof Promise;
+                (await mapped).forEach(c => this.push(c));
+                callback();
+            } catch (err) {
+                if (isPromise) {
+                    // Calling the callback asynchronously with an error wouldn't emit the error, so emit directly
+                    this.emit("error", err);
+                    callback();
+                } else {
+                    callback(err);
+                }
+            }
+        },
+    });
+}
+
+/**
+ * Return a ReadWrite stream that splits streamed chunks using the given separator
+ * @param separator The separator to split by, defaulting to "\n"
+ */
+export function split(separator: string = "\n"): NodeJS.ReadWriteStream {
+    let buffered: string = "";
+    return new Transform({
+        readableObjectMode: true,
+        writableObjectMode: true,
+        async transform(chunk, encoding, callback) {
+            const splitted = chunk.split(separator);
+            if (buffered.length > 0 && splitted.length > 1) {
+                splitted[0] = buffered.concat(splitted[0]);
+                buffered = "";
+            }
+            buffered += splitted[splitted.length - 1];
+            splitted.slice(0, -1).forEach((part: string) => this.push(part));
+            callback();
+        },
+        flush(callback) {
+            callback(undefined, buffered);
+        },
+    });
+}
+
+/**
+ * Return a ReadWrite stream that joins streamed chunks using the given separator
+ * @param separator The separator to join with
+ */
+export function join(separator: string): NodeJS.ReadWriteStream {
+    let isFirstChunk = true;
+    return new Transform({
+        readableObjectMode: true,
+        writableObjectMode: true,
+        async transform(chunk, encoding, callback) {
+            if (!isFirstChunk) {
+                this.push(separator);
+            }
+            this.push(chunk);
+            isFirstChunk = false;
+            callback();
+        },
+    });
+}
+
+/**
+ * Return a ReadWrite stream that collects streamed chunks into an array or buffer
  * @param options
  * @param options.objectMode Whether this stream should behave as a stream of objects
  */