refactor: extract reusable ('@sveltejs/kit/adapter').streamFileContent util for server read

Author: vnphanquang

packages/adapter-cloudflare/src/worker.js

@@ -1,6 +1,7 @@
 import { Server } from 'SERVER';
 import { manifest, prerendered, base_path } from 'MANIFEST';
 import * as Cache from 'worktop/cfw.cache';
+import { streamFileContent } from '@sveltejs/kit/adapter';
 
 const server = new Server(manifest);
 
@@ -21,31 +22,7 @@ export default {
 			// @ts-expect-error env contains environment variables and bindings
 			env,
 			read: (file) =>
-				new ReadableStream({
-					async start(controller) {
-						try {
-							const response = await env.ASSETS.fetch(new URL('/' + file, req.url));
-							if (!response.ok) {
-								throw new Error(`Failed to fetch (${response.status} - ${response.statusText})`);
-							}
-							const reader = response.body.getReader();
-							/** @returns {Promise<void>} */
-							async function pump() {
-								const { done, value } = await reader.read();
-								if (done) {
-									controller.close();
-									return;
-								}
-								controller.enqueue(value);
-								return pump();
-							}
-							return pump();
-						} catch (error) {
-							console.error(`Error reading file ${file}:`, error);
-							controller.error(error);
-						}
-					}
-				})
+				streamFileContent({ fetch: env.ASSETS.fetch, url: new URL('/' + file, req.url) })
 		});
 
 		// skip cache if "cache-control: no-cache" in request

packages/kit/package.json

@@ -83,6 +83,10 @@
 			"types": "./types/index.d.ts",
 			"import": "./src/exports/index.js"
 		},
+		"./adapter": {
+			"types": "./types/index.d.ts",
+			"import": "./src/exports/adapter/index.js"
+		},
 		"./node": {
 			"types": "./types/index.d.ts",
 			"import": "./src/exports/node/index.js"

packages/kit/scripts/generate-dts.js

@@ -5,6 +5,7 @@ await createBundle({
 	output: 'types/index.d.ts',
 	modules: {
 		'@sveltejs/kit': 'src/exports/public.d.ts',
+		'@sveltejs/kit/adapter': 'src/exports/adapter/index.js',
 		'@sveltejs/kit/hooks': 'src/exports/hooks/index.js',
 		'@sveltejs/kit/node': 'src/exports/node/index.js',
 		'@sveltejs/kit/node/polyfills': 'src/exports/node/polyfills.js',

packages/kit/src/exports/adapter/index.js

@@ -0,0 +1,44 @@
+/**
+ * @typedef StreamFileContentOptions
+ * @property {typeof fetch} fetch The fetch function to use for fetching the asset.
+ * @property {string | URL} url The URL of the asset to fetch.
+ * @property {AbortController} [controller] An optional AbortController to cancel the fetch operation.
+ */
+
+/**
+ * synchronously returns a ReadableStream containing the body of an asynchronously fetched asset
+ * original use case: adapters' server read implementation
+ * @param {StreamFileContentOptions } options
+ * @returns {ReadableStream}
+ */
+export function streamFileContent(options) {
+	const { fetch, url, controller: abortController = new AbortController() } = options;
+
+	return new ReadableStream({
+		async start(controller) {
+			try {
+				const response = await fetch(new URL(url), { signal: abortController.signal });
+				if (!response.ok) {
+					throw new Error(`Failed to fetch (${response.status} - ${response.statusText})`);
+				}
+				if (!response.body) {
+					controller.close();
+					return;
+				}
+				const reader = response.body.getReader();
+				while (true) {
+					const { done, value } = await reader.read();
+					if (done) break;
+					controller.enqueue(value);
+				}
+
+				controller.close();
+			} catch (error) {
+				controller.error(error);
+			}
+		},
+		cancel(reason) {
+			abortController.abort(reason);
+		}
+	});
+}

packages/kit/src/exports/adapter/index.spec.js

@@ -0,0 +1,75 @@
+import { test, expect, vi } from 'vitest';
+import { streamFileContent } from './index.js';
+
+const mockedFetch = vi.fn(fetch);
+
+test('stream successfully', async () => {
+	const content = 'Hello, world!';
+	mockedFetch.mockReturnValueOnce(
+		new Promise((resolve) =>
+			resolve(new Response(content, { status: 200, headers: { 'Content-Type': 'text/plain' } }))
+		)
+	);
+	const stream = streamFileContent({
+		fetch: mockedFetch,
+		url: 'http://assets.local/file.txt'
+	});
+	expect(await new Response(stream).text()).toBe(content);
+});
+
+test('stream with 404', async () => {
+	mockedFetch.mockReturnValueOnce(
+		new Promise((resolve) => resolve(new Response(null, { status: 404, statusText: 'Not Found' })))
+	);
+	const stream = streamFileContent({
+		fetch: mockedFetch,
+		url: 'http://assets.local/missing.txt'
+	});
+	await expect(new Response(stream).text()).rejects.toThrow('404 - Not Found');
+});
+
+test('stream with null body', async () => {
+	mockedFetch.mockReturnValueOnce(
+		new Promise((resolve) => resolve(new Response(null, { status: 204, statusText: 'No Content' })))
+	);
+	const stream = streamFileContent({
+		fetch: mockedFetch,
+		url: 'http://assets.local/empty.txt'
+	});
+	expect(await new Response(stream).text()).toBe('');
+});
+
+test('stream with invalid URL', async () => {
+	const stream = streamFileContent({
+		fetch: mockedFetch,
+		url: '/assets.local/invalid.txt'
+	});
+	await expect(new Response(stream).text()).rejects.toThrow('Invalid URL');
+});
+
+test('stream with aborted fetch', async () => {
+	const controller = new AbortController();
+	// abort with DOMException otherwise will fail in CI
+	// see https://github.com/nodejs/node/issues/49557
+	controller.abort(new DOMException('Timeout'));
+	const stream = streamFileContent({
+		fetch: mockedFetch,
+		url: 'http://assets.local/abort.txt',
+		controller
+	});
+	await expect(new Response(stream).text()).rejects.toThrow('Timeout');
+});
+
+test('cancelled stream should trigger abort signal', async () => {
+	const controller = new AbortController();
+	mockedFetch.mockReturnValueOnce(new Promise(() => {}));
+	const stream = streamFileContent({
+		fetch: mockedFetch,
+		url: 'http://assets.local/cancel.txt',
+		controller
+	});
+	const abortListener = vi.fn();
+	controller.signal.addEventListener('abort', abortListener);
+	await stream.cancel('User cancelled');
+	expect(abortListener).toHaveBeenCalled();
+});

packages/kit/types/index.d.ts

@@ -2056,6 +2056,30 @@ declare module '@sveltejs/kit' {
 	export {};
 }
 
+declare module '@sveltejs/kit/adapter' {
+	/**
+	 * synchronously returns a ReadableStream containing the body of an asynchronously fetched asset
+	 * original use case: adapters' server read implementation
+	 * */
+	export function streamFileContent(options: StreamFileContentOptions): ReadableStream;
+	export type StreamFileContentOptions = {
+		/**
+		 * The fetch function to use for fetching the asset.
+		 */
+		fetch: typeof fetch;
+		/**
+		 * The URL of the asset to fetch.
+		 */
+		url: string | URL;
+		/**
+		 * An optional AbortController to cancel the fetch operation.
+		 */
+		controller?: AbortController | undefined;
+	};
+
+	export {};
+}
+
 declare module '@sveltejs/kit/hooks' {
 	/**
 	 * A helper function for sequencing multiple `handle` calls in a middleware-like manner.