feat: backpressure support for concat

ilteoood · Dec 1, 2023 · ce7a97b · ce7a97b
1 parent c7cf8e1
commit ce7a97b
Showing 1 changed file with 13 additions and 15 deletions.
diff --git a/src/concat.ts b/src/concat.ts
@@ -1,24 +1,22 @@
 export const concat = (...readableStreams: ReadableStream[]) => {
 	const fallbackedStreams = readableStreams ?? [];
-	const readableStreamsLength = fallbackedStreams.length;
+	let currentReaderIndex = 0;
+	let currentReader = fallbackedStreams[currentReaderIndex]?.getReader();
 
 	return new ReadableStream({
-		async start(controller) {
-			for (let i = 0; i < readableStreamsLength; i++) {
-				const reader = fallbackedStreams[i].getReader();
-
-				while (true) {
-					const readResult = await reader.read();
+		async pull(controller) {
+			if (!currentReader) {
+				controller.close();
+				return;
+			}
 
-					if (readResult.done) {
-						break;
-					}
+			const readResult = await currentReader.read();
 
-					controller.enqueue(readResult.value);
-				}
+			if (readResult.done) {
+				currentReader = fallbackedStreams[++currentReaderIndex]?.getReader();
+			} else {
+				controller.enqueue(readResult.value);
 			}
-
-			controller.close();
-		},
+		}
 	});
 };