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

catch connection resets in streamables and other error handling improvements

Browse source
This commit is contained in:
wukko 2023-03-31 11:20:49 +06:00
parent 19087944f7
commit 49e85efe23
2 changed files with 45 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
src/cobalt.js
View file

@ -38,6 +38,7 @@ if (fs.existsSync('./.env') && process.env.selfURL && process.env.streamSalt &&
keyGenerator: (req, res) => sha256(req.ip.replace('::ffff:', ''), process.env.streamSalt), keyGenerator: (req, res) => sha256(req.ip.replace('::ffff:', ''), process.env.streamSalt),
handler: (req, res, next, opt) => { handler: (req, res, next, opt) => {
res.status(429).json({ "status": "error", "text": loc(languageCode(req), 'ErrorRateLimit') }); res.status(429).json({ "status": "error", "text": loc(languageCode(req), 'ErrorRateLimit') });
return;
} }
}); });
const apiLimiterStream = rateLimit({ const apiLimiterStream = rateLimit({
@ -48,6 +49,7 @@ if (fs.existsSync('./.env') && process.env.selfURL && process.env.streamSalt &&
keyGenerator: (req, res) => sha256(req.ip.replace('::ffff:', ''), process.env.streamSalt), keyGenerator: (req, res) => sha256(req.ip.replace('::ffff:', ''), process.env.streamSalt),
handler: (req, res, next, opt) => { handler: (req, res, next, opt) => {
res.status(429).json({ "status": "error", "text": loc(languageCode(req), 'ErrorRateLimit') }); res.status(429).json({ "status": "error", "text": loc(languageCode(req), 'ErrorRateLimit') });
return;
} }
}); });
@ -73,10 +75,17 @@ if (fs.existsSync('./.env') && process.env.selfURL && process.env.streamSalt &&
try { try {
JSON.parse(buf); JSON.parse(buf);
if (buf.length > 720) throw new Error(); if (buf.length > 720) throw new Error();
if (String(req.header('Content-Type')) !== "application/json") res.status(500).json({ 'status': 'error', 'text': 'invalid content type header' }) if (String(req.header('Content-Type')) !== "application/json") {
if (String(req.header('Accept')) !== "application/json") res.status(500).json({ 'status': 'error', 'text': 'invalid accept header' }) res.status(400).json({ 'status': 'error', 'text': 'invalid content type header' });
return;
}
if (String(req.header('Accept')) !== "application/json") {
res.status(400).json({ 'status': 'error', 'text': 'invalid accept header' });
return;
}
} catch(e) { } catch(e) {
res.status(500).json({ 'status': 'error', 'text': 'invalid json body.' }) res.status(400).json({ 'status': 'error', 'text': 'invalid json body.' });
return;
} }
} }
})); }));
@ -100,8 +109,10 @@ if (fs.existsSync('./.env') && process.env.selfURL && process.env.streamSalt &&
j = apiJSON(0, { t: loc(lang, 'ErrorCantProcess') }); j = apiJSON(0, { t: loc(lang, 'ErrorCantProcess') });
} }
res.status(j.status).json(j.body); res.status(j.status).json(j.body);
return;
} catch (e) { } catch (e) {
res.status(500).json({ 'status': 'error', 'text': loc(languageCode(req), 'ErrorCantProcess') }) res.status(500).json({ 'status': 'error', 'text': loc(languageCode(req), 'ErrorCantProcess') });
return;
} }
}); });
@ -112,11 +123,13 @@ if (fs.existsSync('./.env') && process.env.selfURL && process.env.streamSalt &&
case 'stream': case 'stream':
if (req.query.p) { if (req.query.p) {
res.status(200).json({ "status": "continue" }); res.status(200).json({ "status": "continue" });
return;
} else if (req.query.t && req.query.h && req.query.e) { } else if (req.query.t && req.query.h && req.query.e) {
stream(res, ip, req.query.t, req.query.h, req.query.e); stream(res, ip, req.query.t, req.query.h, req.query.e);
} else { } else {
let j = apiJSON(0, { t: "no stream id" }) let j = apiJSON(0, { t: "no stream id" })
res.status(j.status).json(j.body); res.status(j.status).json(j.body);
return;
} }
break; break;
case 'onDemand': case 'onDemand':
@ -144,7 +157,8 @@ if (fs.existsSync('./.env') && process.env.selfURL && process.env.streamSalt &&
break; break;
} }
} catch (e) { } catch (e) {
res.status(500).json({ 'status': 'error', 'text': loc(languageCode(req), 'ErrorCantProcess') }) res.status(500).json({ 'status': 'error', 'text': loc(languageCode(req), 'ErrorCantProcess') });
return;
} }
}); });
app.get("/api", (req, res) => { app.get("/api", (req, res) => {

40
src/modules/stream/types.js
View file

@ -16,19 +16,22 @@ export function streamDefault(streamInfo, res) {
isStream: true isStream: true
}); });
stream.pipe(res).on('error', () => { stream.pipe(res).on('error', () => {
res.end(); res.destroy();
}); });
stream.on('error', () => { stream.on('error', () => {
res.end(); res.destroy();
});
stream.on('aborted', () => {
res.destroy();
}); });
} catch (e) { } catch (e) {
res.end(); res.destroy();
} }
} }
export function streamLiveRender(streamInfo, res) { export function streamLiveRender(streamInfo, res) {
try { try {
if (streamInfo.urls.length !== 2) { if (streamInfo.urls.length !== 2) {
res.end(); res.destroy();
return; return;
} }
let audio = got.get(streamInfo.urls[1], { isStream: true }); let audio = got.get(streamInfo.urls[1], { isStream: true });
@ -54,15 +57,24 @@ export function streamLiveRender(streamInfo, res) {
res.setHeader('Content-Disposition', `attachment; filename="${streamInfo.filename}"`); res.setHeader('Content-Disposition', `attachment; filename="${streamInfo.filename}"`);
res.on('error', () => { res.on('error', () => {
ffmpegProcess.kill(); ffmpegProcess.kill();
res.end(); res.destroy();
}); });
ffmpegProcess.stdio[4].pipe(res).on('error', () => { ffmpegProcess.stdio[4].pipe(res).on('error', () => {
ffmpegProcess.kill(); ffmpegProcess.kill();
res.end(); res.destroy();
});; });
audio.pipe(ffmpegProcess.stdio[3]).on('error', () => { audio.pipe(ffmpegProcess.stdio[3]).on('error', () => {
ffmpegProcess.kill(); ffmpegProcess.kill();
res.end(); res.destroy();
});
audio.on('error', () => {
ffmpegProcess.kill();
res.destroy();
});
audio.on('aborted', () => {
ffmpegProcess.kill();
res.destroy();
}); });
ffmpegProcess.on('disconnect', () => ffmpegProcess.kill()); ffmpegProcess.on('disconnect', () => ffmpegProcess.kill());
@ -72,11 +84,11 @@ export function streamLiveRender(streamInfo, res) {
res.on('close', () => ffmpegProcess.kill()); res.on('close', () => ffmpegProcess.kill());
ffmpegProcess.on('error', () => { ffmpegProcess.on('error', () => {
ffmpegProcess.kill(); ffmpegProcess.kill();
res.end(); res.destroy();
}); });
} catch (e) { } catch (e) {
res.end(); res.destroy();
} }
} }
export function streamAudioOnly(streamInfo, res) { export function streamAudioOnly(streamInfo, res) {
@ -119,10 +131,10 @@ export function streamAudioOnly(streamInfo, res) {
res.on('close', () => ffmpegProcess.kill()); res.on('close', () => ffmpegProcess.kill());
ffmpegProcess.on('error', () => { ffmpegProcess.on('error', () => {
ffmpegProcess.kill(); ffmpegProcess.kill();
res.end(); res.destroy();
}); });
} catch (e) { } catch (e) {
res.end(); res.destroy();
} }
} }
export function streamVideoOnly(streamInfo, res) { export function streamVideoOnly(streamInfo, res) {
@ -154,9 +166,9 @@ export function streamVideoOnly(streamInfo, res) {
res.on('close', () => ffmpegProcess.kill()); res.on('close', () => ffmpegProcess.kill());
ffmpegProcess.on('error', () => { ffmpegProcess.on('error', () => {
ffmpegProcess.kill(); ffmpegProcess.kill();
res.end(); res.destroy();
}); });
} catch (e) { } catch (e) {
res.end(); res.destroy();
} }
} }