From b06b1fa39d06f92f13fb80ed3366213bd5f15000 Mon Sep 17 00:00:00 2001
From: Matteo Pietro Dazzi <matteopietro.dazzi@gmail.com>
Date: Sun, 26 Nov 2023 19:22:40 +0100
Subject: [PATCH] feat: map stream

---
 src/fromArray.ts     | 22 +++++++++++++--------
 src/map.ts           | 10 ++++++++++
 src/toArray.ts       | 10 +++++-----
 test/map.test.ts     | 42 ++++++++++++++++++++++++++++++++++++++++
 test/toArray.test.ts | 46 ++++++++++++++++++++++----------------------
 5 files changed, 94 insertions(+), 36 deletions(-)
 create mode 100644 src/map.ts
 create mode 100644 test/map.test.ts

diff --git a/src/fromArray.ts b/src/fromArray.ts
index 4f02c73..a4fd133 100644
--- a/src/fromArray.ts
+++ b/src/fromArray.ts
@@ -1,11 +1,17 @@
-export const fromArray = <T>(sourceArray: T[], strategy?: QueuingStrategy): ReadableStream<T> => {
+export const fromArray = <T>(
+	sourceArray: T[],
+	strategy?: QueuingStrategy,
+): ReadableStream<T> => {
 	const arrayLength = sourceArray.length;
-	return new ReadableStream<T>({
-		start(controller) {
-			for (let i = 0; i < arrayLength; i++) {
-				controller.enqueue(sourceArray[i]);
-			}
-			controller.close();
+	return new ReadableStream<T>(
+		{
+			start(controller) {
+				for (let i = 0; i < arrayLength; i++) {
+					controller.enqueue(sourceArray[i]);
+				}
+				controller.close();
+			},
 		},
-	}, strategy);
+		strategy,
+	);
 };
diff --git a/src/map.ts b/src/map.ts
new file mode 100644
index 0000000..b916593
--- /dev/null
+++ b/src/map.ts
@@ -0,0 +1,10 @@
+type CallbackFn<T, U> = (value: T, index: number) => U;
+
+export const map = <T, U>(callbackfn: CallbackFn<T, U>) => {
+	let index = 0;
+	return new TransformStream<T, U>({
+		transform(chunk, controller) {
+			controller.enqueue(callbackfn(chunk, index++));
+		},
+	});
+};
diff --git a/src/toArray.ts b/src/toArray.ts
index 984788f..a806271 100644
--- a/src/toArray.ts
+++ b/src/toArray.ts
@@ -1,7 +1,7 @@
 export const toArray = <T>(destinationArray: T[]): WritableStream<T> => {
-    return new WritableStream<T>({
-        write(chunk) {
-            destinationArray.push(chunk);
-        },
-    })
+	return new WritableStream<T>({
+		write(chunk) {
+			destinationArray.push(chunk);
+		},
+	});
 };
diff --git a/test/map.test.ts b/test/map.test.ts
new file mode 100644
index 0000000..7b12fda
--- /dev/null
+++ b/test/map.test.ts
@@ -0,0 +1,42 @@
+import { describe, expect, it } from "vitest";
+import { fromArray } from "../src/fromArray";
+import { map } from "../src/map";
+import { toArray } from "../src/toArray";
+
+describe("map", () => {
+	it("should map a stream", async () => {
+		const sourceArray = [1, 2, 3];
+		const destinationArray = [];
+
+		const mapStream = map<number, number>((value) => value * 2);
+
+		await fromArray(sourceArray)
+			.pipeThrough(mapStream)
+			.pipeTo(toArray(destinationArray));
+
+		expect(destinationArray).toEqual([2, 4, 6]);
+	});
+
+	it("should not break for an empty array", async () => {
+		const sourceArray = [];
+		const destinationArray = [];
+
+		const mapStream = map<number, number>((value) => value);
+		await fromArray(sourceArray)
+			.pipeThrough(mapStream)
+			.pipeTo(toArray(destinationArray));
+
+		expect(destinationArray).toEqual([]);
+	});
+
+	it("should throw if parameter is undefined", async () => {
+		// @ts-expect-error undefined parameter for test
+		const mapStream = map<number, number>(undefined);
+
+		expect(
+			fromArray([1])
+				.pipeThrough(mapStream)
+				.pipeTo(toArray([] as number[])),
+		).rejects.toThrow("callbackfn is not a function");
+	});
+});
diff --git a/test/toArray.test.ts b/test/toArray.test.ts
index e551e50..ec06666 100644
--- a/test/toArray.test.ts
+++ b/test/toArray.test.ts
@@ -3,33 +3,33 @@ import { fromArray } from "../src/fromArray";
 import { toArray } from "../src/toArray";
 
 describe("toArray", () => {
-    it('should accumulate a stream into an array', async () => {
-        const sourceArray = [1, 2, 3];
-        const destinationArray = [];
+	it("should accumulate a stream into an array", async () => {
+		const sourceArray = [1, 2, 3];
+		const destinationArray = [];
 
-        await fromArray(sourceArray).pipeTo(toArray(destinationArray));
+		await fromArray(sourceArray).pipeTo(toArray(destinationArray));
 
-        expect(destinationArray).toEqual(sourceArray);
-    })
+		expect(destinationArray).toEqual(sourceArray);
+	});
 
-    it('should not break for an empty array', async () => {
-        const sourceArray = [];
-        const destinationArray = [];
+	it("should not break for an empty array", async () => {
+		const sourceArray = [];
+		const destinationArray = [];
 
-        await fromArray(sourceArray).pipeTo(toArray(destinationArray));
+		await fromArray(sourceArray).pipeTo(toArray(destinationArray));
 
-        expect(destinationArray).toEqual(sourceArray);
-    })
+		expect(destinationArray).toEqual(sourceArray);
+	});
 
-    it('should throw if parameter is undefined', async () => {
-        // @ts-expect-error undefined parameter for test
-        expect(fromArray([1]).pipeTo(toArray(undefined))).rejects.toThrow(
-            "Cannot read properties of undefined (reading 'push')",
-        );
-    })
+	it("should throw if parameter is undefined", async () => {
+		// @ts-expect-error undefined parameter for test
+		expect(fromArray([1]).pipeTo(toArray(undefined))).rejects.toThrow(
+			"Cannot read properties of undefined (reading 'push')",
+		);
+	});
 
-    it('should not throw if source stream is empty', async () => {
-        // @ts-expect-error undefined parameter for test
-        expect(fromArray([]).pipeTo(toArray(undefined))).resolves.toBeUndefined()
-    })
-})
\ No newline at end of file
+	it("should not throw if source stream is empty", async () => {
+		// @ts-expect-error undefined parameter for test
+		expect(fromArray([]).pipeTo(toArray(undefined))).resolves.toBeUndefined();
+	});
+});