blitzwing/cobalt-7
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

stream/internal: refactor, abort controller in more places (#583)

Browse source
This commit is contained in:
jj 2024-06-23 14:51:36 +02:00 committed by GitHub
parent 7815838751
commit a6733ef0cc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 18 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
src/modules/stream/internal.js
View file

@ -39,6 +39,7 @@ async function* readChunks(streamInfo, size) {
async function handleYoutubeStream(streamInfo, res) { async function handleYoutubeStream(streamInfo, res) {
const { signal } = streamInfo.controller; const { signal } = streamInfo.controller;
const cleanup = () => (res.end(), streamInfo.controller.abort());
try { try {
const req = await fetch(streamInfo.url, { const req = await fetch(streamInfo.url, {
@ -52,7 +53,7 @@ async function handleYoutubeStream(streamInfo, res) {
const size = BigInt(req.headers.get('content-length')); const size = BigInt(req.headers.get('content-length'));
if (req.status !== 200 || !size) { if (req.status !== 200 || !size) {
return res.end(); return cleanup();
} }
const generator = readChunks(streamInfo, size); const generator = readChunks(streamInfo, size);
@ -71,17 +72,15 @@ async function handleYoutubeStream(streamInfo, res) {
if (headerValue) res.setHeader(headerName, headerValue); if (headerValue) res.setHeader(headerName, headerValue);
} }
pipe(stream, res, () => res.end()); pipe(stream, res, cleanup);
} catch { } catch {
signal.abort(); cleanup();
res.end();
} }
} }
export async function internalStream(streamInfo, res) { async function handleGenericStream(streamInfo, res) {
if (streamInfo.service === 'youtube') { const { signal } = streamInfo.controller;
return handleYoutubeStream(streamInfo, res); const cleanup = () => (res.end(), streamInfo.controller.abort());
}
try { try {
const req = await request(streamInfo.url, { const req = await request(streamInfo.url, {
@ -90,7 +89,7 @@ export async function internalStream(streamInfo, res) {
host: undefined host: undefined
}, },
dispatcher: streamInfo.dispatcher, dispatcher: streamInfo.dispatcher,
signal: streamInfo.controller.signal, signal,
maxRedirections: 16 maxRedirections: 16
}); });
@ -105,9 +104,17 @@ export async function internalStream(streamInfo, res) {
if (isHlsRequest(req)) { if (isHlsRequest(req)) {
await handleHlsPlaylist(streamInfo, req, res); await handleHlsPlaylist(streamInfo, req, res);
} else { } else {
pipe(req.body, res, () => res.end()); pipe(req.body, res, cleanup);
} }
} catch { } catch {
streamInfo.controller.abort(); cleanup();
} }
} }
export function internalStream(streamInfo, res) {
if (streamInfo.service === 'youtube') {
return handleYoutubeStream(streamInfo, res);
}
return handleGenericStream(streamInfo, res);
}