Add fetchFn to UrlSourceOptions, allow Request as UrlSource input, pass error message to getRetryDelay, add fallback for Blob.stream()

This commit is contained in:
Vanilagy
2025-09-22 11:10:39 +02:00
parent 63145c9ced
commit 15db099caf
3 changed files with 91 additions and 32 deletions
+22
View File
@@ -492,6 +492,9 @@ type UrlSourceOptions = {
// The maximum number of bytes the cache is allowed to hold // The maximum number of bytes the cache is allowed to hold
// in memory. Defaults to 8 MiB. // in memory. Defaults to 8 MiB.
maxCacheSize?: number; maxCacheSize?: number;
// Used to provide a custom fetch function
fetchFn?: typeof fetch;
}; };
``` ```
@@ -516,6 +519,25 @@ const source = new UrlSource('https://example.com/bigbuckbunny.mp4', {
Not setting `getRetryDelay` will default to an infinite, capped exponential backoff pattern. Not setting `getRetryDelay` will default to an infinite, capped exponential backoff pattern.
---
Use `fetchFn` to provide a custom fetch function, usually for polyfill reasons. For example, React Native's `fetch` does not support streamable response bodies, a feature that `UrlSource` requires. In this case, you could use [Expo's `fetch` function](https://docs.expo.dev/versions/latest/sdk/expo/#expofetch-api) instead:
```ts
import { UrlSource } from 'mediabunny';
import { fetch } from 'expo/fetch';
const source = new UrlSource('https://example.com/bigbuckbunny.mp4', {
fetchFn: (input, init) => {
if (typeof input !== 'string') {
// Expo requires string URLs
throw new Error('Expected a string URL.');
}
return fetch(input, init);
},
});
```
### `FilePathSource` ### `FilePathSource`
This input source can be used to load data directly from a file, given a file path. It requires a server-side environment such as Node, Bun, or Deno. This input source can be used to load data directly from a file, given a file path. It requires a server-side environment such as Node, Bun, or Deno.
+5 -4
View File
@@ -575,18 +575,19 @@ const normalizeHeaders = (headers: HeadersInit): Record<string, string> => {
}; };
export const retriedFetch = async ( export const retriedFetch = async (
url: string | URL, fetchFn: typeof fetch,
url: string | URL | Request,
requestInit: RequestInit, requestInit: RequestInit,
getRetryDelay: (previousAttempts: number) => number | null, getRetryDelay: (previousAttempts: number, error: unknown) => number | null,
) => { ) => {
let attempts = 0; let attempts = 0;
while (true) { while (true) {
try { try {
return await fetch(url, requestInit); return await fetchFn(url, requestInit);
} catch (error) { } catch (error) {
attempts++; attempts++;
const retryDelayInSeconds = getRetryDelay(attempts); const retryDelayInSeconds = getRetryDelay(attempts, error);
if (retryDelayInSeconds === null) { if (retryDelayInSeconds === null) {
throw error; throw error;
+54 -18
View File
@@ -200,18 +200,26 @@ export class BlobSource extends Source {
} }
/** @internal */ /** @internal */
_readers = new WeakMap<ReadWorker, ReadableStreamDefaultReader<Uint8Array>>(); _readers = new WeakMap<ReadWorker, ReadableStreamDefaultReader<Uint8Array> | null>();
/** @internal */ /** @internal */
private async _runWorker(worker: ReadWorker) { private async _runWorker(worker: ReadWorker) {
let reader = this._readers.get(worker); let reader = this._readers.get(worker);
if (!reader) { if (reader === undefined) {
if ('stream' in this._blob) {
// Get a reader of the blob starting at the required offset, and then keep it around // Get a reader of the blob starting at the required offset, and then keep it around
reader = this._blob.slice(worker.currentPos).stream().getReader(); const slice = this._blob.slice(worker.currentPos);
reader = slice.stream().getReader();
} else {
// We'll need to use more primitive ways
reader = null;
}
this._readers.set(worker, reader); this._readers.set(worker, reader);
} }
while (worker.currentPos < worker.targetPos && !worker.aborted) { while (worker.currentPos < worker.targetPos && !worker.aborted) {
if (reader) {
const { done, value } = await reader.read(); const { done, value } = await reader.read();
if (done) { if (done) {
this._orchestrator.forgetWorker(worker); this._orchestrator.forgetWorker(worker);
@@ -225,6 +233,12 @@ export class BlobSource extends Source {
this.onread?.(worker.currentPos, worker.currentPos + value.length); this.onread?.(worker.currentPos, worker.currentPos + value.length);
this._orchestrator.supplyWorkerData(worker, value); this._orchestrator.supplyWorkerData(worker, value);
} else {
const data = await this._blob.slice(worker.currentPos, worker.targetPos).arrayBuffer();
this.onread?.(worker.currentPos, worker.currentPos + data.byteLength);
this._orchestrator.supplyWorkerData(worker, new Uint8Array(data));
}
} }
worker.running = false; worker.running = false;
@@ -237,6 +251,8 @@ export class BlobSource extends Source {
} }
const URL_SOURCE_MIN_LOAD_AMOUNT = 0.5 * 2 ** 20; // 0.5 MiB const URL_SOURCE_MIN_LOAD_AMOUNT = 0.5 * 2 ** 20; // 0.5 MiB
const DEFAULT_RETRY_DELAY
= (previousAttempts => Math.min(2 ** (previousAttempts - 2), 16)) satisfies UrlSourceOptions['getRetryDelay'];
/** /**
* Options for {@link UrlSource}. * Options for {@link UrlSource}.
@@ -252,14 +268,21 @@ export type UrlSourceOptions = {
/** /**
* A function that returns the delay (in seconds) before retrying a failed request. The function is called * A function that returns the delay (in seconds) before retrying a failed request. The function is called
* with the number of previous, unsuccessful attempts. If the function returns `null`, no more retries will be made. * with the number of previous, unsuccessful attempts, as well as with the error with which the previous request
* failed. If the function returns `null`, no more retries will be made.
* *
* By default, it uses an exponential backoff algorithm that never fully gives up. * By default, it uses an exponential backoff algorithm that never fully gives up.
*/ */
getRetryDelay?: (previousAttempts: number) => number | null; getRetryDelay?: (previousAttempts: number, error: unknown) => number | null;
/** The maximum number of bytes the cache is allowed to hold in memory. Defaults to 64 MiB. */ /** The maximum number of bytes the cache is allowed to hold in memory. Defaults to 64 MiB. */
maxCacheSize?: number; maxCacheSize?: number;
/**
* A WHATWG-compatible fetch function. You can use this field to polyfill the `fetch` function, add missing
* features, or use a custom implementation.
*/
fetchFn?: typeof fetch;
}; };
/** /**
@@ -270,9 +293,9 @@ export type UrlSourceOptions = {
*/ */
export class UrlSource extends Source { export class UrlSource extends Source {
/** @internal */ /** @internal */
_url: URL; _url: string | URL | Request;
/** @internal */ /** @internal */
_getRetryDelay: (previousAttempts: number) => number | null; _getRetryDelay: (previousAttempts: number, error: unknown) => number | null;
/** @internal */ /** @internal */
_options: UrlSourceOptions; _options: UrlSourceOptions;
/** @internal */ /** @internal */
@@ -285,11 +308,15 @@ export class UrlSource extends Source {
/** Creates a new {@link UrlSource} backed by the resource at the specified URL. */ /** Creates a new {@link UrlSource} backed by the resource at the specified URL. */
constructor( constructor(
url: string | URL, url: string | URL | Request,
options: UrlSourceOptions = {}, options: UrlSourceOptions = {},
) { ) {
if (typeof url !== 'string' && !(url instanceof URL)) { if (
throw new TypeError('url must be a string or URL.'); typeof url !== 'string'
&& !(url instanceof URL)
&& !(typeof Request !== 'undefined' && url instanceof Request)
) {
throw new TypeError('url must be a string, URL or Request.');
} }
if (!options || typeof options !== 'object') { if (!options || typeof options !== 'object') {
throw new TypeError('options must be an object.'); throw new TypeError('options must be an object.');
@@ -306,14 +333,16 @@ export class UrlSource extends Source {
) { ) {
throw new TypeError('options.maxCacheSize, when provided, must be a non-negative integer.'); throw new TypeError('options.maxCacheSize, when provided, must be a non-negative integer.');
} }
if (options.fetchFn !== undefined && typeof options.fetchFn !== 'function') {
throw new TypeError('options.fetchFn, when provided, must be a function.');
// Won't bother validating this function beyond this
}
super(); super();
this._url = url instanceof URL this._url = url;
? url
: new URL(url, typeof location !== 'undefined' ? location.href : undefined);
this._options = options; this._options = options;
this._getRetryDelay = options.getRetryDelay ?? (previousAttempts => Math.min(2 ** (previousAttempts - 2), 8)); this._getRetryDelay = options.getRetryDelay ?? DEFAULT_RETRY_DELAY;
this._orchestrator = new ReadOrchestrator({ this._orchestrator = new ReadOrchestrator({
maxCacheSize: options.maxCacheSize ?? (64 * 2 ** 20 /* 64 MiB */), maxCacheSize: options.maxCacheSize ?? (64 * 2 ** 20 /* 64 MiB */),
@@ -334,6 +363,7 @@ export class UrlSource extends Source {
const abortController = new AbortController(); const abortController = new AbortController();
const response = await retriedFetch( const response = await retriedFetch(
this._options.fetchFn ?? fetch,
this._url, this._url,
mergeRequestInit(this._options.requestInit ?? {}, { mergeRequestInit(this._options.requestInit ?? {}, {
headers: { headers: {
@@ -347,7 +377,8 @@ export class UrlSource extends Source {
); );
if (!response.ok) { if (!response.ok) {
throw new Error(`Error fetching ${this._url}: ${response.status} ${response.statusText}`); // eslint-disable-next-line @typescript-eslint/no-base-to-string
throw new Error(`Error fetching ${String(this._url)}: ${response.status} ${response.statusText}`);
} }
let worker: ReadWorker; let worker: ReadWorker;
@@ -401,6 +432,7 @@ export class UrlSource extends Source {
if (!abortController) { if (!abortController) {
abortController = new AbortController(); abortController = new AbortController();
response = await retriedFetch( response = await retriedFetch(
this._options.fetchFn ?? fetch,
this._url, this._url,
mergeRequestInit(this._options.requestInit ?? {}, { mergeRequestInit(this._options.requestInit ?? {}, {
headers: { headers: {
@@ -415,7 +447,8 @@ export class UrlSource extends Source {
assert(response); assert(response);
if (!response.ok) { if (!response.ok) {
throw new Error(`Error fetching ${this._url}: ${response.status} ${response.statusText}`); // eslint-disable-next-line @typescript-eslint/no-base-to-string
throw new Error(`Error fetching ${String(this._url)}: ${response.status} ${response.statusText}`);
} }
if (worker.currentPos > 0 && response.status !== 206) { if (worker.currentPos > 0 && response.status !== 206) {
@@ -434,7 +467,10 @@ export class UrlSource extends Source {
} }
if (!response.body) { if (!response.body) {
throw new Error('Missing HTTP response body.'); throw new Error(
'Missing HTTP response body stream. The used fetch function must provide the response body as a'
+ ' ReadableStream.',
);
} }
const reader = response.body.getReader(); const reader = response.body.getReader();
@@ -452,7 +488,7 @@ export class UrlSource extends Source {
try { try {
readResult = await reader.read(); readResult = await reader.read();
} catch (error) { } catch (error) {
const retryDelayInSeconds = this._getRetryDelay(1); const retryDelayInSeconds = this._getRetryDelay(1, error);
if (retryDelayInSeconds !== null) { if (retryDelayInSeconds !== null) {
console.error('Error while reading response stream. Attempting to resume.', error); console.error('Error while reading response stream. Attempting to resume.', error);
await new Promise(resolve => setTimeout(resolve, 1000 * retryDelayInSeconds)); await new Promise(resolve => setTimeout(resolve, 1000 * retryDelayInSeconds));