454 lines
13 KiB
JavaScript
454 lines
13 KiB
JavaScript
// RTMP stream session management and ffmpeg probe orchestration.
|
|
|
|
const crypto = require('crypto');
|
|
const fs = require('fs');
|
|
const path = require('path');
|
|
const { spawn } = require('child_process');
|
|
|
|
function createRtmpStreamService(options) {
|
|
const mediaDir = options && options.mediaDir ? options.mediaDir : null;
|
|
const ffmpegPath = options && options.ffmpegPath ? options.ffmpegPath : 'ffmpeg';
|
|
const ffprobePath = options && options.ffprobePath ? options.ffprobePath : 'ffprobe';
|
|
const probeIntervalMs = options && options.probeIntervalMs ? Number(options.probeIntervalMs) : 2000;
|
|
const sessions = new Map();
|
|
const watchedSources = new Set();
|
|
const sourceStatusCache = new Map();
|
|
const sourceProbePromises = new Map();
|
|
let probeTimer = null;
|
|
|
|
if (!mediaDir) {
|
|
throw new Error('mediaDir is required');
|
|
}
|
|
|
|
const cacheRoot = path.join(mediaDir, 'rtmp-cache');
|
|
|
|
function normalizeSourceUrl(value) {
|
|
const sourceUrl = String(value || '').trim();
|
|
if (!sourceUrl || !/^rtmps?:\/\//i.test(sourceUrl)) {
|
|
const error = new Error('A valid RTMP URL is required.');
|
|
error.statusCode = 400;
|
|
throw error;
|
|
}
|
|
return sourceUrl;
|
|
}
|
|
|
|
function getSessionKey(sourceUrl, disableAudio) {
|
|
return crypto.createHash('sha1').update(String(sourceUrl)).update('\0').update(disableAudio ? '1' : '0').digest('hex');
|
|
}
|
|
|
|
async function ensureDirectory(dirPath) {
|
|
await fs.promises.mkdir(dirPath, { recursive: true });
|
|
}
|
|
|
|
async function waitForFile(filePath, timeoutMs) {
|
|
const startedAt = Date.now();
|
|
while (Date.now() - startedAt < timeoutMs) {
|
|
try {
|
|
const stat = await fs.promises.stat(filePath);
|
|
if (stat.isFile() && stat.size > 0) {
|
|
return true;
|
|
}
|
|
} catch (_error) {
|
|
// keep waiting
|
|
}
|
|
await new Promise((resolve) => setTimeout(resolve, 100));
|
|
}
|
|
return false;
|
|
}
|
|
|
|
function runCommand(command, args, timeoutMs) {
|
|
return new Promise(function (resolve) {
|
|
var child = spawn(command, args, {
|
|
stdio: ['ignore', 'pipe', 'pipe']
|
|
});
|
|
var stdout = '';
|
|
var stderr = '';
|
|
var finished = false;
|
|
var timer = null;
|
|
|
|
function done(result) {
|
|
if (finished) {
|
|
return;
|
|
}
|
|
finished = true;
|
|
if (timer) {
|
|
clearTimeout(timer);
|
|
timer = null;
|
|
}
|
|
resolve(result);
|
|
}
|
|
|
|
child.stdout.on('data', function (chunk) {
|
|
stdout += String(chunk || '');
|
|
});
|
|
|
|
child.stderr.on('data', function (chunk) {
|
|
stderr += String(chunk || '');
|
|
});
|
|
|
|
child.on('error', function (error) {
|
|
done({ ok: false, error: error, stdout: stdout, stderr: stderr, timedOut: false });
|
|
});
|
|
|
|
child.on('exit', function (code, signal) {
|
|
done({ ok: code === 0, code: code, signal: signal, stdout: stdout, stderr: stderr, timedOut: false });
|
|
});
|
|
|
|
timer = setTimeout(function () {
|
|
try {
|
|
child.kill('SIGKILL');
|
|
} catch (_error) {
|
|
// ignore timeout cleanup errors
|
|
}
|
|
done({ ok: false, code: null, signal: 'SIGKILL', stdout: stdout, stderr: stderr, timedOut: true });
|
|
}, Math.max(1000, Number(timeoutMs || 0) || 4000));
|
|
});
|
|
}
|
|
|
|
async function probeSourceUrl(sourceUrl) {
|
|
const normalizedSource = normalizeSourceUrl(sourceUrl);
|
|
const result = await runCommand(ffprobePath, [
|
|
'-hide_banner',
|
|
'-loglevel', 'error',
|
|
'-rw_timeout', '3000000',
|
|
'-show_streams',
|
|
'-of', 'json',
|
|
normalizedSource
|
|
], 4000);
|
|
|
|
if (!result.ok) {
|
|
return {
|
|
live: false,
|
|
timedOut: Boolean(result.timedOut),
|
|
stderr: String(result.stderr || '').trim()
|
|
};
|
|
}
|
|
|
|
try {
|
|
const parsed = JSON.parse(String(result.stdout || '{}'));
|
|
const streams = Array.isArray(parsed && parsed.streams) ? parsed.streams : [];
|
|
return {
|
|
live: streams.length > 0,
|
|
timedOut: false,
|
|
stderr: String(result.stderr || '').trim()
|
|
};
|
|
} catch (_error) {
|
|
return {
|
|
live: false,
|
|
timedOut: false,
|
|
stderr: String(result.stderr || '').trim()
|
|
};
|
|
}
|
|
}
|
|
|
|
function getSourceStatus(sourceUrl) {
|
|
return sourceStatusCache.get(String(sourceUrl || '').trim()) || null;
|
|
}
|
|
|
|
function setSourceStatus(sourceUrl, nextStatus) {
|
|
const normalizedSource = String(sourceUrl || '').trim();
|
|
if (!normalizedSource) {
|
|
return null;
|
|
}
|
|
const status = Object.assign({
|
|
live: false,
|
|
probing: false,
|
|
checkedAt: Date.now(),
|
|
timedOut: false,
|
|
stderr: ''
|
|
}, nextStatus || {});
|
|
status.live = Boolean(status.live);
|
|
status.probing = Boolean(status.probing);
|
|
status.checkedAt = Number(status.checkedAt || Date.now());
|
|
sourceStatusCache.set(normalizedSource, status);
|
|
return status;
|
|
}
|
|
|
|
function registerSourceWatch(sourceUrl) {
|
|
const normalizedSource = normalizeSourceUrl(sourceUrl);
|
|
watchedSources.add(normalizedSource);
|
|
return normalizedSource;
|
|
}
|
|
|
|
async function refreshSourceStatus(sourceUrl) {
|
|
const normalizedSource = registerSourceWatch(sourceUrl);
|
|
const existingPromise = sourceProbePromises.get(normalizedSource) || null;
|
|
if (existingPromise) {
|
|
return existingPromise;
|
|
}
|
|
|
|
setSourceStatus(normalizedSource, Object.assign({}, getSourceStatus(normalizedSource) || {}, {
|
|
probing: true
|
|
}));
|
|
|
|
const probePromise = probeSourceUrl(normalizedSource).then(function (probe) {
|
|
return setSourceStatus(normalizedSource, {
|
|
live: Boolean(probe.live),
|
|
probing: false,
|
|
checkedAt: Date.now(),
|
|
timedOut: Boolean(probe.timedOut),
|
|
stderr: String(probe.stderr || '')
|
|
});
|
|
}).catch(function (error) {
|
|
return setSourceStatus(normalizedSource, {
|
|
live: false,
|
|
probing: false,
|
|
checkedAt: Date.now(),
|
|
timedOut: false,
|
|
stderr: String(error && error.message ? error.message : '')
|
|
});
|
|
}).finally(function () {
|
|
sourceProbePromises.delete(normalizedSource);
|
|
});
|
|
|
|
sourceProbePromises.set(normalizedSource, probePromise);
|
|
return probePromise;
|
|
}
|
|
|
|
function scheduleSourceRefresh(sourceUrl) {
|
|
const normalizedSource = registerSourceWatch(sourceUrl);
|
|
const currentStatus = getSourceStatus(normalizedSource);
|
|
if (currentStatus && currentStatus.probing) {
|
|
return sourceProbePromises.get(normalizedSource) || Promise.resolve(currentStatus);
|
|
}
|
|
return refreshSourceStatus(normalizedSource);
|
|
}
|
|
|
|
function isSourceStatusFresh(status) {
|
|
if (!status || !status.checkedAt) {
|
|
return false;
|
|
}
|
|
return Date.now() - Number(status.checkedAt || 0) < probeIntervalMs;
|
|
}
|
|
|
|
function sweepWatchedSources() {
|
|
watchedSources.forEach(function (sourceUrl) {
|
|
const status = getSourceStatus(sourceUrl);
|
|
if (!status || !isSourceStatusFresh(status) || status.live === false) {
|
|
scheduleSourceRefresh(sourceUrl);
|
|
}
|
|
});
|
|
}
|
|
|
|
if (probeIntervalMs > 0) {
|
|
probeTimer = setInterval(function () {
|
|
sweepWatchedSources();
|
|
}, probeIntervalMs);
|
|
if (probeTimer && typeof probeTimer.unref === 'function') {
|
|
probeTimer.unref();
|
|
}
|
|
}
|
|
|
|
async function isSessionLive(session) {
|
|
if (!session || !session.process) {
|
|
return false;
|
|
}
|
|
if (session.exitCode !== undefined && session.exitCode !== null) {
|
|
return false;
|
|
}
|
|
if (session.exitSignal !== undefined && session.exitSignal !== null) {
|
|
return false;
|
|
}
|
|
try {
|
|
const stat = await fs.promises.stat(session.manifestPath);
|
|
return stat.isFile() && stat.size > 0;
|
|
} catch (_error) {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
function buildPlaylistUrl(key) {
|
|
return '/api/rtmp/streams/' + encodeURIComponent(key) + '/index.m3u8';
|
|
}
|
|
|
|
async function ensureSession(sourceUrl, disableAudio) {
|
|
const normalizedSource = normalizeSourceUrl(sourceUrl);
|
|
const normalizedDisableAudio = Boolean(disableAudio);
|
|
const key = getSessionKey(normalizedSource, normalizedDisableAudio);
|
|
const directory = path.join(cacheRoot, key);
|
|
const manifestPath = path.join(directory, 'index.m3u8');
|
|
|
|
const existingSession = sessions.get(key) || null;
|
|
if (existingSession && existingSession.process && existingSession.exitCode === null && existingSession.exitSignal === null) {
|
|
return existingSession;
|
|
}
|
|
|
|
if (existingSession) {
|
|
sessions.delete(key);
|
|
try {
|
|
if (existingSession.process && typeof existingSession.process.kill === 'function') {
|
|
existingSession.process.kill('SIGKILL');
|
|
}
|
|
} catch (_error) {
|
|
// ignore cleanup errors
|
|
}
|
|
try {
|
|
await fs.promises.rm(directory, { recursive: true, force: true });
|
|
} catch (_error2) {
|
|
// ignore cleanup errors
|
|
}
|
|
}
|
|
|
|
await ensureDirectory(directory);
|
|
|
|
const args = [
|
|
'-hide_banner',
|
|
'-loglevel', 'warning',
|
|
'-nostdin',
|
|
'-i', normalizedSource,
|
|
'-fflags', '+genpts',
|
|
'-f', 'hls',
|
|
'-hls_time', '1',
|
|
'-hls_init_time', '1',
|
|
'-hls_list_size', '8',
|
|
'-hls_flags', 'delete_segments+append_list+omit_endlist+independent_segments+program_date_time',
|
|
'-hls_segment_filename', path.join(directory, 'segment-%05d.ts')
|
|
];
|
|
|
|
if (normalizedDisableAudio) {
|
|
args.push('-an');
|
|
args.push('-c:v', 'copy');
|
|
} else {
|
|
args.push('-c', 'copy');
|
|
}
|
|
|
|
args.push(manifestPath);
|
|
|
|
const child = spawn(ffmpegPath, args, {
|
|
stdio: ['ignore', 'ignore', 'pipe']
|
|
});
|
|
|
|
child.stderr.on('data', function (chunk) {
|
|
const message = String(chunk || '').trim();
|
|
if (message) {
|
|
console.error('[rtmp]', message);
|
|
}
|
|
});
|
|
|
|
child.on('exit', function (code, signal) {
|
|
const session = sessions.get(key);
|
|
if (session && session.process === child) {
|
|
session.process = null;
|
|
session.exitCode = code;
|
|
session.exitSignal = signal;
|
|
}
|
|
});
|
|
|
|
const session = {
|
|
key: key,
|
|
sourceUrl: normalizedSource,
|
|
disableAudio: normalizedDisableAudio,
|
|
directory: directory,
|
|
manifestPath: manifestPath,
|
|
playlistUrl: buildPlaylistUrl(key),
|
|
process: child,
|
|
exitCode: null,
|
|
exitSignal: null,
|
|
ready: waitForFile(manifestPath, 5000)
|
|
};
|
|
|
|
sessions.set(key, session);
|
|
return session;
|
|
}
|
|
|
|
async function getPlaylistUrl(sourceUrl, disableAudio) {
|
|
const session = await ensureSession(sourceUrl, disableAudio);
|
|
await session.ready.catch(function () {
|
|
return false;
|
|
});
|
|
return session.playlistUrl + '?t=' + Date.now();
|
|
}
|
|
|
|
async function getSessionStatus(sourceUrl, disableAudio) {
|
|
const normalizedSource = normalizeSourceUrl(sourceUrl);
|
|
const normalizedDisableAudio = Boolean(disableAudio);
|
|
registerSourceWatch(normalizedSource);
|
|
|
|
let sourceStatus = getSourceStatus(normalizedSource);
|
|
if (!sourceStatus) {
|
|
scheduleSourceRefresh(normalizedSource).catch(function (_error) {
|
|
return null;
|
|
});
|
|
sourceStatus = getSourceStatus(normalizedSource);
|
|
} else if (!isSourceStatusFresh(sourceStatus) && !sourceStatus.probing) {
|
|
scheduleSourceRefresh(normalizedSource).catch(function (_error) {
|
|
return null;
|
|
});
|
|
}
|
|
|
|
if (!sourceStatus || !sourceStatus.live) {
|
|
return {
|
|
session: null,
|
|
ready: false,
|
|
live: false,
|
|
probing: Boolean(sourceStatus && sourceStatus.probing),
|
|
checkedAt: sourceStatus ? sourceStatus.checkedAt : null,
|
|
timedOut: Boolean(sourceStatus && sourceStatus.timedOut),
|
|
stderr: String(sourceStatus && sourceStatus.stderr || '')
|
|
};
|
|
}
|
|
|
|
const session = await ensureSession(normalizedSource, normalizedDisableAudio);
|
|
const ready = await session.ready.catch(function () {
|
|
return false;
|
|
});
|
|
const live = ready && Boolean(session && session.process && session.exitCode === null && session.exitSignal === null);
|
|
return {
|
|
session: session,
|
|
ready: ready,
|
|
live: live,
|
|
probing: Boolean(sourceStatus && sourceStatus.probing),
|
|
checkedAt: sourceStatus ? sourceStatus.checkedAt : null,
|
|
timedOut: false,
|
|
stderr: ''
|
|
};
|
|
}
|
|
|
|
function getSessionByKey(key) {
|
|
return sessions.get(String(key || '').trim()) || null;
|
|
}
|
|
|
|
async function getManifestFilePath(key) {
|
|
const session = getSessionByKey(key);
|
|
if (!session) {
|
|
return null;
|
|
}
|
|
if (!session.process || session.exitCode !== null || session.exitSignal !== null) {
|
|
return null;
|
|
}
|
|
const live = await isSessionLive(session);
|
|
if (!live) {
|
|
return null;
|
|
}
|
|
return session.manifestPath;
|
|
}
|
|
|
|
async function getSegmentFilePath(key, fileName) {
|
|
const session = getSessionByKey(key);
|
|
if (!session) {
|
|
return null;
|
|
}
|
|
const segmentName = path.basename(String(fileName || '').trim());
|
|
if (!segmentName || segmentName === 'index.m3u8') {
|
|
return null;
|
|
}
|
|
return path.join(session.directory, segmentName);
|
|
}
|
|
|
|
return {
|
|
ensureSession: ensureSession,
|
|
getPlaylistUrl: getPlaylistUrl,
|
|
getSessionByKey: getSessionByKey,
|
|
isSessionLive: isSessionLive,
|
|
getSessionStatus: getSessionStatus,
|
|
refreshSourceStatus: refreshSourceStatus,
|
|
scheduleSourceRefresh: scheduleSourceRefresh,
|
|
getSourceStatus: getSourceStatus,
|
|
getManifestFilePath: getManifestFilePath,
|
|
getSegmentFilePath: getSegmentFilePath
|
|
};
|
|
}
|
|
|
|
module.exports = {
|
|
createRtmpStreamService: createRtmpStreamService
|
|
}; |