mirror of
https://github.com/fancive/obsidian-parallel-reader.git
synced 2026-07-22 06:53:43 +00:00
feat: add streaming timeout protection
Add streamingTimeoutMs field to PluginSettings (default 120000ms). In streamingFetch(), use Promise.race between the stream reader and a timeout promise that aborts the AbortController and throws a clear timeout error. External cancellation signals are forwarded to the same controller so only one abort path is needed. https://claude.ai/code/session_016QvEfqw6YZ3RjwBHrJ4w8S
This commit is contained in:
parent
dbb8ea4a5b
commit
e7384fd875
3 changed files with 53 additions and 13 deletions
|
|
@ -41,6 +41,7 @@ export const DEFAULT_SETTINGS: PluginSettings = {
|
|||
exportFolder: 'Reading/Articles',
|
||||
cliTimeoutMs: 120000,
|
||||
streaming: true,
|
||||
streamingTimeoutMs: 120000,
|
||||
};
|
||||
|
||||
export const API_FORMATS: Record<string, ApiFormat> = {
|
||||
|
|
|
|||
|
|
@ -70,24 +70,15 @@ export interface StreamProgress {
|
|||
done: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform a streaming fetch with SSE parsing.
|
||||
* Uses the native Fetch API (available in Electron/Obsidian).
|
||||
* Returns the full accumulated text when done.
|
||||
*/
|
||||
export async function streamingFetch(
|
||||
async function doStreamingFetch(
|
||||
url: string,
|
||||
headers: Record<string, string>,
|
||||
body: unknown,
|
||||
extractDelta: DeltaExtractor,
|
||||
onProgress?: (progress: StreamProgress) => void,
|
||||
signal?: AbortSignal,
|
||||
settings?: PluginSettings | null,
|
||||
onProgress: ((progress: StreamProgress) => void) | undefined,
|
||||
signal: AbortSignal,
|
||||
settings: PluginSettings | null | undefined,
|
||||
): Promise<string> {
|
||||
if (typeof globalThis.fetch !== 'function') {
|
||||
throw new Error('Streaming requires fetch API');
|
||||
}
|
||||
|
||||
const response = await globalThis.fetch(url, {
|
||||
method: 'POST',
|
||||
headers,
|
||||
|
|
@ -138,3 +129,50 @@ export async function streamingFetch(
|
|||
onProgress?.({ accumulated, done: true });
|
||||
return accumulated;
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform a streaming fetch with SSE parsing and configurable timeout.
|
||||
* Uses the native Fetch API (available in Electron/Obsidian).
|
||||
* Returns the full accumulated text when done.
|
||||
*/
|
||||
export async function streamingFetch(
|
||||
url: string,
|
||||
headers: Record<string, string>,
|
||||
body: unknown,
|
||||
extractDelta: DeltaExtractor,
|
||||
onProgress?: (progress: StreamProgress) => void,
|
||||
signal?: AbortSignal,
|
||||
settings?: PluginSettings | null,
|
||||
): Promise<string> {
|
||||
if (typeof globalThis.fetch !== 'function') {
|
||||
throw new Error('Streaming requires fetch API');
|
||||
}
|
||||
|
||||
const timeoutMs = settings?.streamingTimeoutMs ?? 120000;
|
||||
const timeoutController = new AbortController();
|
||||
|
||||
if (signal) {
|
||||
if (signal.aborted) {
|
||||
timeoutController.abort();
|
||||
} else {
|
||||
signal.addEventListener('abort', () => timeoutController.abort(), { once: true });
|
||||
}
|
||||
}
|
||||
|
||||
let timeoutId: ReturnType<typeof setTimeout> | null = null;
|
||||
const timeoutPromise = new Promise<never>((_, reject) => {
|
||||
timeoutId = setTimeout(() => {
|
||||
timeoutController.abort();
|
||||
reject(new Error(`Streaming timed out after ${timeoutMs}ms`));
|
||||
}, timeoutMs);
|
||||
});
|
||||
|
||||
try {
|
||||
return await Promise.race([
|
||||
doStreamingFetch(url, headers, body, extractDelta, onProgress, timeoutController.signal, settings),
|
||||
timeoutPromise,
|
||||
]);
|
||||
} finally {
|
||||
if (timeoutId !== null) clearTimeout(timeoutId);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -66,6 +66,7 @@ export interface PluginSettings {
|
|||
exportFolder: string;
|
||||
cliTimeoutMs: number;
|
||||
streaming: boolean;
|
||||
streamingTimeoutMs: number;
|
||||
}
|
||||
|
||||
/* ---------- Provider types ---------- */
|
||||
|
|
|
|||
Loading…
Reference in a new issue