Compare commits

..

1 Commits

Author SHA1 Message Date
OTYAK
3b1362ebbd Merge 5d60ab1139 into 7955d8e408 2025-03-29 17:31:50 +01:00

View File

@@ -146,10 +146,7 @@ export const POST = async (req: Request) => {
} }
} catch (error) { } catch (error) {
reject( reject(
Response.json( Response.json({ message: 'Error parsing data' }, { status: 500 }),
{ message: 'Error parsing data' },
{ status: 500 },
),
); );
} }
}); });
@@ -160,10 +157,7 @@ export const POST = async (req: Request) => {
emitter.on('error', (error: any) => { emitter.on('error', (error: any) => {
reject( reject(
Response.json( Response.json({ message: 'Search error', error }, { status: 500 }),
{ message: 'Search error', error },
{ status: 500 },
),
); );
}); });
}, },
@@ -172,6 +166,7 @@ export const POST = async (req: Request) => {
const encoder = new TextEncoder(); const encoder = new TextEncoder();
// Create an AbortController to handle cancellation
const abortController = new AbortController(); const abortController = new AbortController();
const { signal } = abortController; const { signal } = abortController;
@@ -179,48 +174,43 @@ export const POST = async (req: Request) => {
start(controller) { start(controller) {
let sources: any[] = []; let sources: any[] = [];
controller.enqueue( // Send an initial message to keep the connection alive
encoder.encode( controller.enqueue(encoder.encode(JSON.stringify({
JSON.stringify({ type: 'init',
type: 'init', data: 'Stream connected'
data: 'Stream connected', }) + '\n'));
}) + '\n',
),
);
// Set up cleanup function for when client disconnects
signal.addEventListener('abort', () => { signal.addEventListener('abort', () => {
// Remove all listeners from emitter to prevent memory leaks
emitter.removeAllListeners(); emitter.removeAllListeners();
// Close the controller if it's still active
try { try {
controller.close(); controller.close();
} catch (error) {} } catch (error) {
// Controller might already be closed
}
}); });
emitter.on('data', (data: string) => { emitter.on('data', (data: string) => {
// Check if request has been cancelled before processing
if (signal.aborted) return; if (signal.aborted) return;
try { try {
const parsedData = JSON.parse(data); const parsedData = JSON.parse(data);
if (parsedData.type === 'response') { if (parsedData.type === 'response') {
controller.enqueue( controller.enqueue(encoder.encode(JSON.stringify({
encoder.encode( type: 'response',
JSON.stringify({ data: parsedData.data
type: 'response', }) + '\n'));
data: parsedData.data,
}) + '\n',
),
);
} else if (parsedData.type === 'sources') { } else if (parsedData.type === 'sources') {
sources = parsedData.data; sources = parsedData.data;
controller.enqueue( controller.enqueue(encoder.encode(JSON.stringify({
encoder.encode( type: 'sources',
JSON.stringify({ data: sources
type: 'sources', }) + '\n'));
data: sources,
}) + '\n',
),
);
} }
} catch (error) { } catch (error) {
controller.error(error); controller.error(error);
@@ -228,19 +218,17 @@ export const POST = async (req: Request) => {
}); });
emitter.on('end', () => { emitter.on('end', () => {
// Check if request has been cancelled before processing
if (signal.aborted) return; if (signal.aborted) return;
controller.enqueue( controller.enqueue(encoder.encode(JSON.stringify({
encoder.encode( type: 'done'
JSON.stringify({ }) + '\n'));
type: 'done',
}) + '\n',
),
);
controller.close(); controller.close();
}); });
emitter.on('error', (error: any) => { emitter.on('error', (error: any) => {
// Check if request has been cancelled before processing
if (signal.aborted) return; if (signal.aborted) return;
controller.error(error); controller.error(error);
@@ -248,14 +236,14 @@ export const POST = async (req: Request) => {
}, },
cancel() { cancel() {
abortController.abort(); abortController.abort();
}, }
}); });
return new Response(stream, { return new Response(stream, {
headers: { headers: {
'Content-Type': 'text/event-stream', 'Content-Type': 'application/json',
'Cache-Control': 'no-cache, no-transform', 'Cache-Control': 'no-cache, no-transform',
Connection: 'keep-alive', 'Connection': 'keep-alive',
}, },
}); });
} catch (err: any) { } catch (err: any) {