mirror of
https://github.com/bilawalsidhu/gods-eye-view.git
synced 2026-11-05 19:04:02 +00:00
fix(cctv): bound HLS sessions and isolate consumer playback leases
This commit is contained in:
Bilawal Sidhu committed
1 parent
48666cb9a4
commit
292cd610f3
22 files changed
+665
-730
No files matched your search
+41
-122
@@ -1,5 +1,3 @@
|
||||
import path from 'node:path';
|
||||
import { promises as fsp } from 'node:fs';
|
||||
import { createCctvCatalog } from './cctv/catalog.js';
|
||||
import {
|
||||
normalizeFeedType,
|
||||
@@ -19,7 +17,7 @@ import {
|
||||
CCTV_MAX_SOURCES_CEILING,
|
||||
} from './cctv/constants.js';
|
||||
import { sanitizeCctvRangeHeader } from './cctv/range.js';
|
||||
import { createHlsPuller, createHlsRemuxer } from './cctv/stream.js';
|
||||
import { createHlsPuller } from './cctv/stream.js';
|
||||
import { googleServerApiKey } from './places/google-key.js';
|
||||
export { CCTV_FRAME_FETCH_TIMEOUT_MS, fetchCctvImageFromUpstream };
|
||||
/**
|
||||
@@ -45,7 +43,6 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
|
||||
const HEALTH_MAX_ENTRIES = CCTV_MAX_SOURCES_CEILING;
|
||||
/** Live HLS strategies (see ./cctv/stream.js). Shared across dev and preview. */
|
||||
const puller = createHlsPuller();
|
||||
const remuxer = createHlsRemuxer();
|
||||
|
||||
/** Update the health entry for a camera, evicting the oldest entry if at capacity. */
|
||||
const setHealth = (cameraId, patch) => {
|
||||
@@ -136,7 +133,6 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
|
||||
server.httpServer?.on('close', () => {
|
||||
puller.shutdown();
|
||||
|
||||
remuxer.shutdown();
|
||||
});
|
||||
server.middlewares.use('/api/cctv', async (req, res) => {
|
||||
try {
|
||||
@@ -205,133 +201,56 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
|
||||
}
|
||||
|
||||
if (url.pathname.startsWith('/media/')) {
|
||||
// /media/<id> is the playlist (or the whole media for non-HLS feeds);
|
||||
|
||||
// /media/<id>/<file> is a playlist- or segment-relative sub-path.
|
||||
|
||||
const mediaPath = url.pathname.replace('/media/', '');
|
||||
|
||||
const slash = mediaPath.indexOf('/');
|
||||
|
||||
const cameraId =
|
||||
decodeURIComponent(
|
||||
(slash === -1 ? mediaPath : mediaPath.slice(0, slash)).trim(),
|
||||
) || 'camera';
|
||||
|
||||
const rel = slash === -1 ? '' : mediaPath.slice(slash + 1);
|
||||
|
||||
const match = /^\/media\/([^/]+)(?:\/(seg_(\d+)\.ts))?$/.exec(url.pathname);
|
||||
if (!match) { res.writeHead(404); res.end(); return; }
|
||||
const cameraId = decodeURIComponent(match[1]);
|
||||
const source = sourceById.get(cameraId);
|
||||
|
||||
const mediaUrl = source?.url || '';
|
||||
|
||||
const feedType = normalizeFeedType(source?.feedType || 'image');
|
||||
|
||||
// HLS feeds are served from a proxy-owned segment store with a
|
||||
|
||||
// proxy-generated playlist, so upstream session churn never reaches
|
||||
|
||||
// the browser. http(s) .m3u8 upstreams use the Node puller; RTMP and
|
||||
|
||||
// other stream kinds use ffmpeg when it is available. Anything else
|
||||
|
||||
// falls through to the direct proxy below.
|
||||
|
||||
const hlsStrategy =
|
||||
feedType === 'hls' &&
|
||||
mediaUrl &&
|
||||
/^(https?|rtmps?|rtmpt):\/\//i.test(mediaUrl)
|
||||
? /\.m3u8(\?|$)/i.test(mediaUrl)
|
||||
? puller
|
||||
: remuxer.isAvailable()
|
||||
? remuxer
|
||||
: null
|
||||
: null;
|
||||
|
||||
if (hlsStrategy) {
|
||||
const entry = await hlsStrategy.ensure(cameraId, mediaUrl);
|
||||
|
||||
const safe = path.basename(rel || 'out.m3u8');
|
||||
|
||||
const isPlaylist = safe.endsWith('.m3u8');
|
||||
|
||||
const starting = () => {
|
||||
res.writeHead(503, {
|
||||
'Content-Type': 'application/json',
|
||||
|
||||
'Cache-Control': 'no-store',
|
||||
|
||||
'Retry-After': '2',
|
||||
});
|
||||
|
||||
res.end(JSON.stringify({ error: 'Stream starting' }));
|
||||
};
|
||||
|
||||
if (isPlaylist) {
|
||||
if (!(await hlsStrategy.waitReady(entry))) {
|
||||
starting();
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
const playlist = await hlsStrategy.buildPlaylist(entry, cameraId);
|
||||
|
||||
if (!playlist) {
|
||||
starting();
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
setHealth(cameraId, {
|
||||
status: 'ok',
|
||||
|
||||
sourceKind: 'live',
|
||||
|
||||
label: source?.provider || 'Configured source',
|
||||
|
||||
message: 'Live stream connected',
|
||||
});
|
||||
|
||||
res.writeHead(200, {
|
||||
'Content-Type': 'application/vnd.apple.mpegurl',
|
||||
|
||||
'Cache-Control': 'no-store',
|
||||
|
||||
'X-CCTV-Source':
|
||||
hlsStrategy === puller ? 'hls-pull' : 'ffmpeg-remux',
|
||||
});
|
||||
|
||||
res.end(playlist);
|
||||
|
||||
const leaseId = url.searchParams.get('lease');
|
||||
if (feedType === 'hls' && !/^[a-f0-9-]{36}$/i.test(leaseId || '')) { res.writeHead(400); res.end(); return; }
|
||||
if (req.method === 'DELETE') {
|
||||
if (leaseId) puller.release(cameraId, leaseId);
|
||||
res.writeHead(204); res.end(); return;
|
||||
}
|
||||
if (req.method !== 'GET') { res.writeHead(405); res.end(); return; }
|
||||
if (feedType === 'hls') {
|
||||
if (!/^https?:\/\//i.test(mediaUrl) || !/\.m3u8(?:\?|$)/i.test(mediaUrl)) {
|
||||
res.writeHead(503, { 'Content-Type': 'application/json' });
|
||||
res.end(JSON.stringify({ error: 'This stream requires an unsupported transport; use the frame fallback' }));
|
||||
return;
|
||||
}
|
||||
|
||||
if (match[2]) {
|
||||
const body = puller.getSegment(cameraId, url.searchParams.get('session'), Number(match[3]), leaseId);
|
||||
res.writeHead(body ? 200 : 404, { 'Content-Type': 'video/mp2t', 'Cache-Control': 'no-store' });
|
||||
res.end(body || undefined); return;
|
||||
}
|
||||
const downstream = watchDownstreamClose(res);
|
||||
let entry;
|
||||
const cancelPending = () => { if (entry) puller.release(cameraId, leaseId); };
|
||||
try {
|
||||
entry.lastAccess = Date.now();
|
||||
|
||||
const buf = await fsp.readFile(path.join(entry.dir, safe));
|
||||
|
||||
res.writeHead(200, {
|
||||
'Content-Type': 'video/mp2t',
|
||||
|
||||
'Cache-Control': 'no-store',
|
||||
|
||||
'X-CCTV-Source':
|
||||
hlsStrategy === puller ? 'hls-pull' : 'ffmpeg-remux',
|
||||
});
|
||||
|
||||
res.end(buf);
|
||||
entry = await puller.ensure(cameraId, mediaUrl, leaseId);
|
||||
if (downstream.closed) { cancelPending(); return; }
|
||||
downstream.signal.addEventListener('abort', cancelPending, { once: true });
|
||||
if (!(await puller.waitReady(entry, downstream.signal))) throw new Error('Stream unavailable');
|
||||
const playlist = await puller.buildPlaylist(entry, cameraId, leaseId);
|
||||
if (downstream.closed) return;
|
||||
if (!playlist) throw new Error('Stream unavailable');
|
||||
setHealth(cameraId, { status: 'ok', sourceKind: 'live', label: source?.provider || 'Configured source', message: 'Live HLS connected' });
|
||||
res.writeHead(200, { 'Content-Type': 'application/vnd.apple.mpegurl', 'Cache-Control': 'no-store', 'X-CCTV-Source': 'hls-pull', 'X-CCTV-Session': entry.token });
|
||||
res.end(playlist);
|
||||
} catch {
|
||||
res.writeHead(404, {
|
||||
'Content-Type': 'application/json',
|
||||
|
||||
'Cache-Control': 'no-store',
|
||||
});
|
||||
|
||||
res.end(JSON.stringify({ error: 'Segment not found' }));
|
||||
setHealth(cameraId, { status: 'degraded', sourceKind: 'fallback', label: source?.provider || 'Configured source', message: 'Live HLS unavailable' });
|
||||
if (!downstream.closed) {
|
||||
res.writeHead(503, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store', 'Retry-After': '2' });
|
||||
res.end(JSON.stringify({ error: 'Live stream unavailable' }));
|
||||
}
|
||||
} finally {
|
||||
downstream.signal.removeEventListener('abort', cancelPending);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
if (match[2] || req.method !== 'GET') { res.writeHead(404); res.end(); return; }
|
||||
|
||||
if (!mediaUrl || !/^https?:\/\//i.test(mediaUrl)) {
|
||||
setHealth(cameraId, {
|
||||
|
||||
@@ -1603,12 +1603,13 @@ export async function loadDelDOTSourcesFromOpenData() {
|
||||
const resp = await fetch(DELDOT_CCTV_URL, {
|
||||
headers: { Accept: 'application/json' },
|
||||
signal: AbortSignal.timeout(CCTV_SOURCE_FETCH_TIMEOUT_MS),
|
||||
redirect: 'error',
|
||||
});
|
||||
if (!resp.ok) {
|
||||
console.warn('[CCTV] DelDOT source download failed:', resp.status);
|
||||
return [];
|
||||
}
|
||||
const payload = await resp.json();
|
||||
const payload = await readResponseJsonCapped(resp, 2 * 1024 * 1024);
|
||||
const rows = Array.isArray(payload?.videoCameras)
|
||||
? payload.videoCameras
|
||||
: [];
|
||||
@@ -1623,12 +1624,14 @@ export async function loadDelDOTSourcesFromOpenData() {
|
||||
// Delaware bounding box — a bad upstream coord can't place a camera out of state.
|
||||
if (lat < 38.4 || lat > 39.9 || lon < -75.85 || lon > -75.0) continue;
|
||||
|
||||
// Official-host pin: RTMP-over-HTTP (rtmpt:80) is the steady transport.
|
||||
const streamUrl = String(row?.urls?.rtmp || '');
|
||||
if (!streamUrl.startsWith('rtmpt://video.deldot.gov:80/live/')) continue;
|
||||
// Prefer the catalog's HTTPS HLS transport; no subprocess is required.
|
||||
let stream;
|
||||
try { stream = new URL(String(row?.urls?.m3u8s || '')); } catch { continue; }
|
||||
if (stream.origin !== 'https://video.deldot.gov' || stream.username || stream.password || !/^\/live\/[A-Za-z0-9_.-]+\/playlist\.m3u8$/.test(stream.pathname)) continue;
|
||||
const streamUrl = stream.href;
|
||||
|
||||
const id = String(row?.id || '').trim();
|
||||
if (!id) continue;
|
||||
if (!/^[A-Za-z0-9_-]{1,80}$/.test(id)) continue;
|
||||
const cameraId = `deldot-${id.toLowerCase()}`;
|
||||
const title = String(row?.title || '').trim();
|
||||
|
||||
@@ -1657,10 +1660,10 @@ export async function loadDelDOTSourcesFromOpenData() {
|
||||
fovDeg: hasHeading ? 56 : 44,
|
||||
rangeM: hasHeading ? 210 : 145,
|
||||
mountHeightM: hasHeading ? 10 : 8,
|
||||
groundElevationM: 10, // Delaware is near sea level and flat.
|
||||
groundElevationM: 10, // Estimated prior; client ground resolution owns placement.
|
||||
feedType: 'hls',
|
||||
url: streamUrl,
|
||||
snapshotUrl: streamUrl,
|
||||
snapshotUrl: '',
|
||||
sourceKind: 'deldot-open-data',
|
||||
license: 'Public DelDOT traffic camera',
|
||||
});
|
||||
|
||||
+207
-438
@@ -1,469 +1,238 @@
|
||||
/**
|
||||
* Live HLS delivery strategies for the CCTV media proxy.
|
||||
*
|
||||
* Both strategies own a per-camera segment store on disk (seg_<n>.ts in a
|
||||
* temp dir) and generate the playlist the browser sees, so upstream session
|
||||
* behaviour never reaches the client:
|
||||
*
|
||||
* createHlsPuller — for http(s) .m3u8 upstreams. Pure Node, no dependencies.
|
||||
* createHlsRemuxer — for RTMP and other stream kinds, via ffmpeg (-c copy).
|
||||
* Optional: absent, those cameras fall back to stills.
|
||||
*
|
||||
* Both expose the same surface: ensure(cameraId, url) -> entry,
|
||||
* waitReady(entry), buildPlaylist(entry, cameraId), shutdown(). The media
|
||||
* route in ../cctv.js picks a strategy by URL and serves files from entry.dir.
|
||||
*/
|
||||
import os from 'node:os';
|
||||
import path from 'node:path';
|
||||
import { promises as fsp } from 'node:fs';
|
||||
import { spawn } from 'node:child_process';
|
||||
import { CCTV_MEDIA_FETCH_TIMEOUT_MS } from './constants.js';
|
||||
/** Bounded, demand-owned HLS sessions. No subprocess, disk cache or idle sweep. */
|
||||
import { randomUUID } from 'node:crypto';
|
||||
|
||||
/**
|
||||
* Read a text response with a byte cap, cancelling the body on overflow.
|
||||
* media.js keeps its byte-buffer reader private; playlists need text.
|
||||
* @returns {Promise<{tooLarge:boolean,text:string}>}
|
||||
*/
|
||||
export async function readCappedResponseText(upstream, maxBytes) {
|
||||
const declared = Number(upstream.headers.get('content-length'));
|
||||
if (Number.isFinite(declared) && declared > maxBytes) {
|
||||
try { await upstream.body?.cancel(); } catch { /* no-op */ }
|
||||
return { tooLarge: true, text: '' };
|
||||
}
|
||||
if (!upstream.body || typeof upstream.body[Symbol.asyncIterator] !== 'function') {
|
||||
const text = await upstream.text();
|
||||
return text.length > maxBytes ? { tooLarge: true, text: '' } : { tooLarge: false, text };
|
||||
}
|
||||
const decoder = new TextDecoder();
|
||||
let text = '';
|
||||
let total = 0;
|
||||
for await (const chunk of upstream.body) {
|
||||
total += chunk.length;
|
||||
if (total > maxBytes) {
|
||||
try { await upstream.body.cancel(); } catch { /* no-op */ }
|
||||
return { tooLarge: true, text: '' };
|
||||
export const HLS_LIMITS = Object.freeze({
|
||||
sessions: 2,
|
||||
leasesPerSession: 8,
|
||||
playlistBytes: 256 * 1024,
|
||||
segmentBytes: 4 * 1024 * 1024,
|
||||
sessionBytes: 24 * 1024 * 1024,
|
||||
segments: 12,
|
||||
pollMs: 2000,
|
||||
idleMs: 15000,
|
||||
timeoutMs: 10000,
|
||||
readyMs: 12000,
|
||||
});
|
||||
|
||||
/** Refuse redirects; count actual streamed bytes, including chunked responses. */
|
||||
export async function fetchHlsBytes(url, { fetchImpl = fetch, signal, maxBytes, timeoutMs = HLS_LIMITS.timeoutMs } = {}) {
|
||||
const controller = new AbortController();
|
||||
const abort = () => controller.abort();
|
||||
if (signal?.aborted) controller.abort();
|
||||
else signal?.addEventListener('abort', abort, { once: true });
|
||||
const timer = setTimeout(abort, timeoutMs);
|
||||
let reader;
|
||||
let response;
|
||||
try {
|
||||
controller.signal.throwIfAborted();
|
||||
response = await fetchImpl(url, { redirect: 'error', signal: controller.signal,
|
||||
headers: { 'User-Agent': 'gods-eye-view-cctv-proxy/1.0' } });
|
||||
if (!response.ok || response.redirected || (response.url && new URL(response.url).href !== new URL(url).href))
|
||||
throw new Error('HLS upstream refused');
|
||||
if (Number(response.headers.get('content-length')) > maxBytes) throw new Error('HLS response too large');
|
||||
reader = response.body?.getReader?.();
|
||||
if (!reader) throw new Error('HLS response is not streamable');
|
||||
const chunks = [];
|
||||
let length = 0;
|
||||
for (;;) {
|
||||
controller.signal.throwIfAborted();
|
||||
const { done, value } = await reader.read();
|
||||
controller.signal.throwIfAborted();
|
||||
if (done) break;
|
||||
length += value.byteLength;
|
||||
if (length > maxBytes) throw new Error('HLS response too large');
|
||||
chunks.push(Buffer.from(value));
|
||||
}
|
||||
text += decoder.decode(chunk, { stream: true });
|
||||
return Buffer.concat(chunks, length);
|
||||
} finally {
|
||||
controller.abort();
|
||||
if (reader) await reader.cancel().catch(() => {});
|
||||
else await response?.body?.cancel().catch(() => {});
|
||||
reader?.releaseLock();
|
||||
clearTimeout(timer);
|
||||
signal?.removeEventListener('abort', abort);
|
||||
}
|
||||
text += decoder.decode();
|
||||
return { tooLarge: false, text };
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-camera HLS puller with a proxy-owned segment store (no ffmpeg).
|
||||
*
|
||||
* Wowza-style upstreams expire sessions fast, but their segment numbering is
|
||||
* stream-global (EXT-X-MEDIA-SEQUENCE carries across sessions). So the proxy
|
||||
* polls the chunklist every ~2 s (which also keeps the session warm), re-reads
|
||||
* the master the moment a poll is refused, and downloads every segment it has
|
||||
* not yet seen, keyed by that global sequence. Zero overlap, zero loss across
|
||||
* session rotations, exact EXTINF durations from upstream, PTS continuous
|
||||
* because it is the same encoder. Serves the same file layout as
|
||||
* createHlsRemuxer (seg_<n>.ts in a per-camera dir) so the media route is
|
||||
* shared. Idle cameras are reaped after IDLE_MS.
|
||||
*/
|
||||
export function createHlsPuller() {
|
||||
const OUT_ROOT = path.join(os.tmpdir(), 'gev-hls-pull');
|
||||
const IDLE_MS = 60_000;
|
||||
const READY_TIMEOUT_MS = 20_000;
|
||||
const POLL_MS = 2_000;
|
||||
const PLAYLIST_WINDOW = 15;
|
||||
const KEEP_ON_DISK = 24;
|
||||
const MIN_READY_SEGMENTS = 3;
|
||||
const FETCH_TIMEOUT_MS = CCTV_MEDIA_FETCH_TIMEOUT_MS;
|
||||
const MAX_PLAYLIST_BYTES = 256 * 1024;
|
||||
const UA = 'gods-eye-view-cctv-proxy/1.0';
|
||||
/** @type {Map<string, any>} */
|
||||
export function sameOriginHlsUrl(value, base) {
|
||||
const url = new URL(value, base);
|
||||
if (!['https:', 'http:'].includes(url.protocol) || url.origin !== new URL(base).origin || url.username || url.password)
|
||||
throw new Error('HLS reference escapes registered origin');
|
||||
return url.href;
|
||||
}
|
||||
|
||||
/** Only clear MPEG-TS media playlists: encrypted, fMP4 and byte ranges fail closed. */
|
||||
export function parseHlsMedia(text, base, limit = HLS_LIMITS.segments) {
|
||||
if (!text.startsWith('#EXTM3U') || /#EXT-X-(?:KEY|MAP|BYTERANGE|I-FRAMES-ONLY)/.test(text))
|
||||
throw new Error('Unsupported HLS playlist');
|
||||
let seq = 0;
|
||||
let duration = null;
|
||||
let discontinuity = false;
|
||||
const segments = [];
|
||||
for (const raw of text.split('\n')) {
|
||||
const line = raw.trim();
|
||||
if (line.startsWith('#EXT-X-MEDIA-SEQUENCE:')) {
|
||||
seq = Number(line.slice(22));
|
||||
if (!Number.isSafeInteger(seq) || seq < 0) throw new Error('Invalid media sequence');
|
||||
} else if (line === '#EXT-X-DISCONTINUITY') {
|
||||
discontinuity = true;
|
||||
} else if (line.startsWith('#EXTINF:')) {
|
||||
duration = Number.parseFloat(line.slice(8));
|
||||
if (!Number.isFinite(duration) || duration <= 0 || duration > 60) throw new Error('Invalid segment duration');
|
||||
} else if (line && !line.startsWith('#')) {
|
||||
if (duration === null || !Number.isSafeInteger(seq)) throw new Error('Invalid HLS segment');
|
||||
const uri = sameOriginHlsUrl(line, base);
|
||||
if (!new URL(uri).pathname.endsWith('.ts')) throw new Error('Only MPEG-TS segments are supported');
|
||||
segments.push({ seq: seq++, duration, uri, discontinuity });
|
||||
discontinuity = false;
|
||||
duration = null;
|
||||
}
|
||||
}
|
||||
return segments.slice(-limit);
|
||||
}
|
||||
|
||||
export function createHlsPuller({ fetchImpl = fetch, limits = HLS_LIMITS } = {}) {
|
||||
const active = new Map();
|
||||
const headers = { 'User-Agent': UA };
|
||||
const fetchText = async (url) => {
|
||||
const r = await fetch(url, { headers, signal: AbortSignal.timeout(FETCH_TIMEOUT_MS) });
|
||||
if (!r.ok) { const e = new Error(`HTTP ${r.status}`); e.status = r.status; throw e; }
|
||||
const { tooLarge, text } = await readCappedResponseText(r, MAX_PLAYLIST_BYTES);
|
||||
if (tooLarge) throw new Error('playlist too large');
|
||||
return text;
|
||||
let closed = false;
|
||||
const stop = (cameraId, token) => {
|
||||
const entry = active.get(cameraId);
|
||||
if (!entry || (token && token !== entry.token)) return;
|
||||
active.delete(cameraId);
|
||||
entry.stopping = true;
|
||||
entry.controller.abort();
|
||||
clearTimeout(entry.timer);
|
||||
for (const lease of entry.leases.values()) clearTimeout(lease.timer);
|
||||
entry.leases.clear();
|
||||
entry.segments.clear();
|
||||
entry.upstream.clear();
|
||||
entry.bytes = 0;
|
||||
};
|
||||
/** Follow a master playlist to its first media playlist (or return it if already one). */
|
||||
const resolveChunklist = async (masterUrl) => {
|
||||
const text = await fetchText(masterUrl);
|
||||
if (text.includes('#EXTINF')) return { url: masterUrl, text };
|
||||
const first = text.split('\n').map((l) => l.trim()).find((l) => l && !l.startsWith('#'));
|
||||
if (!first) throw new Error('master playlist has no variants');
|
||||
const url = new URL(first, masterUrl);
|
||||
if (url.origin !== new URL(masterUrl).origin) throw new Error('variant escapes origin');
|
||||
return { url: url.toString(), text: await fetchText(url.toString()) };
|
||||
const release = (cameraId, leaseId) => {
|
||||
const entry = active.get(cameraId);
|
||||
const lease = entry?.leases.get(leaseId);
|
||||
if (!lease) return;
|
||||
clearTimeout(lease.timer);
|
||||
entry.leases.delete(leaseId);
|
||||
if (!entry.leases.size) stop(cameraId, entry.token);
|
||||
};
|
||||
/** Parse a media playlist into [{seq, duration, uri}] using MEDIA-SEQUENCE + index. */
|
||||
const parseMedia = (text, baseUrl) => {
|
||||
let mediaSeq = 0;
|
||||
let target = 0;
|
||||
let dur = null;
|
||||
const out = [];
|
||||
for (const raw of text.split('\n')) {
|
||||
const line = raw.trim();
|
||||
if (!line) continue;
|
||||
if (line.startsWith('#EXT-X-MEDIA-SEQUENCE:')) mediaSeq = Number(line.slice(22)) || 0;
|
||||
else if (line.startsWith('#EXT-X-TARGETDURATION:')) target = Number(line.slice(22)) || 0;
|
||||
else if (line.startsWith('#EXTINF:')) dur = parseFloat(line.slice(8)) || null;
|
||||
else if (!line.startsWith('#')) {
|
||||
let uri;
|
||||
try { uri = new URL(line, baseUrl); } catch { continue; }
|
||||
if (uri.origin !== new URL(baseUrl).origin) continue;
|
||||
const m = /_(\d+)\.ts(\?|$)/i.exec(uri.pathname);
|
||||
const seq = m ? Number(m[1]) : mediaSeq + out.length;
|
||||
out.push({ seq, duration: dur || target || 2, uri: uri.toString() });
|
||||
dur = null;
|
||||
}
|
||||
const touch = (entry, leaseId, create = false) => {
|
||||
let lease = entry.leases.get(leaseId);
|
||||
if (!lease) {
|
||||
if (!create || entry.leases.size >= limits.leasesPerSession) throw new Error('HLS lease capacity reached');
|
||||
lease = {};
|
||||
entry.leases.set(leaseId, lease);
|
||||
}
|
||||
return { segments: out, target };
|
||||
clearTimeout(lease.timer);
|
||||
lease.timer = setTimeout(() => release(entry.cameraId, leaseId), limits.idleMs);
|
||||
lease.timer.unref?.();
|
||||
};
|
||||
const downloadSegment = async (entry, seg) => {
|
||||
const r = await fetch(seg.uri, { headers, signal: AbortSignal.timeout(FETCH_TIMEOUT_MS) });
|
||||
if (!r.ok) throw new Error(`segment ${seg.seq} HTTP ${r.status}`);
|
||||
const buf = Buffer.from(await r.arrayBuffer());
|
||||
const file = path.join(entry.dir, `seg_${seg.seq}.ts`);
|
||||
await fsp.writeFile(`${file}.part`, buf);
|
||||
await fsp.rename(`${file}.part`, file);
|
||||
entry.segments.set(seg.seq, { duration: seg.duration });
|
||||
};
|
||||
const prune = (entry) => {
|
||||
const seqs = [...entry.segments.keys()].sort((a, b) => a - b);
|
||||
while (seqs.length > KEEP_ON_DISK) {
|
||||
const n = seqs.shift();
|
||||
entry.segments.delete(n);
|
||||
fsp.rm(path.join(entry.dir, `seg_${n}.ts`), { force: true }).catch(() => {});
|
||||
}
|
||||
};
|
||||
const poll = async (cameraId, entry) => {
|
||||
if (entry.stopping || entry.polling) return;
|
||||
entry.polling = true;
|
||||
const read = (entry, url, maxBytes) => fetchHlsBytes(url, {
|
||||
fetchImpl, signal: entry.controller.signal, maxBytes, timeoutMs: limits.timeoutMs,
|
||||
});
|
||||
const poll = async (entry) => {
|
||||
try {
|
||||
let text;
|
||||
if (!entry.chunklistUrl) {
|
||||
const r = await resolveChunklist(entry.masterUrl);
|
||||
entry.chunklistUrl = r.url;
|
||||
text = r.text;
|
||||
} else {
|
||||
try {
|
||||
text = await fetchText(entry.chunklistUrl);
|
||||
} catch (err) {
|
||||
if (err.status === 403 || err.status === 404) {
|
||||
// Session rotated: re-read the master for the live chunklist.
|
||||
const r = await resolveChunklist(entry.masterUrl);
|
||||
if (r.url !== entry.chunklistUrl) console.info(`[CCTV:pull ${cameraId}] session rotated`);
|
||||
entry.chunklistUrl = r.url;
|
||||
text = r.text;
|
||||
} else {
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
let base = entry.chunklistUrl || entry.url;
|
||||
let text = (await read(entry, base, limits.playlistBytes)).toString('utf8');
|
||||
if (!text.includes('#EXTINF:')) {
|
||||
const variant = text.split('\n').map((line) => line.trim()).find((line) => line && !line.startsWith('#'));
|
||||
if (!variant || !text.includes('#EXT-X-STREAM-INF:')) throw new Error('Missing HLS variant');
|
||||
base = sameOriginHlsUrl(variant, entry.url);
|
||||
text = (await read(entry, base, limits.playlistBytes)).toString('utf8');
|
||||
}
|
||||
const { segments } = parseMedia(text, entry.chunklistUrl);
|
||||
for (const seg of segments) {
|
||||
if (entry.stopping) break;
|
||||
if (entry.segments.has(seg.seq) || entry.inFlight.has(seg.seq)) continue;
|
||||
if (entry.lastSeq !== null && seg.seq <= entry.lastSeq) continue; // older than what we hold
|
||||
entry.inFlight.add(seg.seq);
|
||||
try {
|
||||
await downloadSegment(entry, seg);
|
||||
entry.lastSeq = Math.max(entry.lastSeq ?? -1, seg.seq);
|
||||
} catch (err) {
|
||||
console.warn(`[CCTV:pull ${cameraId}]`, err.message);
|
||||
} finally {
|
||||
entry.inFlight.delete(seg.seq);
|
||||
entry.chunklistUrl = base;
|
||||
const segments = parseHlsMedia(text, base, limits.segments);
|
||||
const newest = segments.at(-1)?.seq ?? -1;
|
||||
const restarted = newest < entry.upstreamNewest || segments.some((segment) => {
|
||||
const prior = entry.upstream.get(segment.seq);
|
||||
return prior && prior !== segment.uri;
|
||||
});
|
||||
if (restarted) {
|
||||
entry.upstream.clear();
|
||||
entry.pendingDiscontinuity = true;
|
||||
}
|
||||
entry.upstreamNewest = newest;
|
||||
for (const segment of segments) {
|
||||
if (entry.stopping) return;
|
||||
if (entry.upstream.has(segment.seq)) continue;
|
||||
const body = await read(entry, segment.uri, limits.segmentBytes);
|
||||
if (entry.stopping) return;
|
||||
// Evict BEFORE retaining the incoming segment. One bounded download may
|
||||
// exist in addition to the cache; no unbounded disk output is possible.
|
||||
while (entry.segments.size && (entry.bytes + body.length > limits.sessionBytes || entry.segments.size >= limits.segments)) {
|
||||
const oldest = entry.segments.keys().next().value;
|
||||
const removed = entry.segments.get(oldest);
|
||||
entry.bytes -= removed.body.length;
|
||||
if (removed.discontinuity) entry.discontinuitiesRemoved++;
|
||||
entry.segments.delete(oldest);
|
||||
}
|
||||
if (body.length > limits.sessionBytes) throw new Error('Segment exceeds session budget');
|
||||
const seq = entry.nextSeq++;
|
||||
entry.segments.set(seq, { ...segment, seq, body, discontinuity: segment.discontinuity || entry.pendingDiscontinuity });
|
||||
entry.pendingDiscontinuity = false;
|
||||
entry.upstream.set(segment.seq, segment.uri);
|
||||
while (entry.upstream.size > limits.segments * 2) entry.upstream.delete(entry.upstream.keys().next().value);
|
||||
entry.bytes += body.length;
|
||||
}
|
||||
prune(entry);
|
||||
entry.failures = 0;
|
||||
} catch (err) {
|
||||
entry.failures += 1;
|
||||
if (entry.failures === 1 || entry.failures % 10 === 0) {
|
||||
console.warn(`[CCTV:pull ${cameraId}] poll failed (${entry.failures}):`, err.message);
|
||||
}
|
||||
if (entry.failures >= 3) entry.chunklistUrl = null; // force master re-resolve
|
||||
} finally {
|
||||
entry.polling = false;
|
||||
}
|
||||
};
|
||||
const ensure = async (cameraId, masterUrl) => {
|
||||
let entry = active.get(cameraId);
|
||||
if (entry) {
|
||||
entry.lastAccess = Date.now();
|
||||
return entry;
|
||||
}
|
||||
const dir = path.join(OUT_ROOT, cameraId.replace(/[^a-zA-Z0-9_-]/g, '_'));
|
||||
await fsp.rm(dir, { recursive: true, force: true }).catch(() => {});
|
||||
await fsp.mkdir(dir, { recursive: true });
|
||||
entry = {
|
||||
dir,
|
||||
masterUrl,
|
||||
chunklistUrl: null,
|
||||
segments: new Map(),
|
||||
inFlight: new Set(),
|
||||
lastSeq: null,
|
||||
lastAccess: Date.now(),
|
||||
failures: 0,
|
||||
polling: false,
|
||||
stopping: false,
|
||||
timer: null,
|
||||
};
|
||||
active.set(cameraId, entry);
|
||||
entry.timer = setInterval(() => { poll(cameraId, entry); }, POLL_MS);
|
||||
entry.timer.unref?.();
|
||||
poll(cameraId, entry);
|
||||
return entry;
|
||||
};
|
||||
const waitReady = async (entry) => {
|
||||
const deadline = Date.now() + READY_TIMEOUT_MS;
|
||||
while (Date.now() < deadline) {
|
||||
if (entry.segments.size >= MIN_READY_SEGMENTS) return true;
|
||||
await new Promise((r) => setTimeout(r, 250));
|
||||
}
|
||||
return false;
|
||||
};
|
||||
const buildPlaylist = async (entry, cameraId) => {
|
||||
const seqs = [...entry.segments.keys()].sort((a, b) => a - b);
|
||||
if (seqs.length < 2) return null;
|
||||
const window = seqs.slice(-PLAYLIST_WINDOW);
|
||||
const prefix = `/api/cctv/media/${encodeURIComponent(cameraId)}/`;
|
||||
const maxDur = Math.max(...window.map((n) => entry.segments.get(n).duration));
|
||||
const lines = [
|
||||
'#EXTM3U',
|
||||
'#EXT-X-VERSION:3',
|
||||
`#EXT-X-TARGETDURATION:${Math.ceil(maxDur)}`,
|
||||
`#EXT-X-MEDIA-SEQUENCE:${window[0]}`,
|
||||
];
|
||||
let prev = null;
|
||||
for (const n of window) {
|
||||
if (prev !== null && n !== prev + 1) lines.push('#EXT-X-DISCONTINUITY');
|
||||
lines.push(`#EXTINF:${entry.segments.get(n).duration.toFixed(3)},`, `${prefix}seg_${n}.ts`);
|
||||
prev = n;
|
||||
}
|
||||
return lines.join('\n') + '\n';
|
||||
};
|
||||
const stopEntry = (entry) => {
|
||||
entry.stopping = true;
|
||||
if (entry.timer) { clearInterval(entry.timer); entry.timer = null; }
|
||||
fsp.rm(entry.dir, { recursive: true, force: true }).catch(() => {});
|
||||
};
|
||||
const sweep = setInterval(() => {
|
||||
const now = Date.now();
|
||||
for (const [id, entry] of active) {
|
||||
if (now - entry.lastAccess > IDLE_MS) { stopEntry(entry); active.delete(id); }
|
||||
}
|
||||
}, 30_000);
|
||||
sweep.unref?.();
|
||||
const shutdown = () => {
|
||||
clearInterval(sweep);
|
||||
for (const [, entry] of active) stopEntry(entry);
|
||||
active.clear();
|
||||
};
|
||||
return { isAvailable: () => true, ensure, waitReady, buildPlaylist, shutdown };
|
||||
}
|
||||
/**
|
||||
* Per-camera ffmpeg segment producer with a proxy-owned playlist.
|
||||
*
|
||||
* Wowza-style upstreams expire HLS sessions fast and irregularly; no single
|
||||
* long-lived reader survives them (ffmpeg's HLS demuxer exits on a master
|
||||
* playlist 403 above the -reconnect layer). So ffmpeg is treated as
|
||||
* disposable: it only writes numbered MPEG-TS segments (-f segment) into a
|
||||
* per-camera dir, and when it dies it is respawned with -segment_start_number
|
||||
* continuing from the last file on disk. The playlist is generated here from
|
||||
* whatever segments exist, with a monotonic media sequence, so respawns are
|
||||
* invisible to hls.js. -c copy: no re-encode, no quality loss. Processes are
|
||||
* reaped after IDLE_MS with no client access; dirs are pruned to a window.
|
||||
*/
|
||||
export function createHlsRemuxer() {
|
||||
const OUT_ROOT = path.join(os.tmpdir(), 'gev-hls');
|
||||
const IDLE_MS = 60_000;
|
||||
const READY_TIMEOUT_MS = 20_000;
|
||||
const SEGMENT_SECONDS = 4;
|
||||
const STALL_MS = 15_000;
|
||||
const PLAYLIST_WINDOW = 15; // segments listed
|
||||
const KEEP_ON_DISK = 24; // segments retained w/ in-flight fetches
|
||||
const MIN_READY_SEGMENTS = 3;
|
||||
const RESPAWN_DELAY_MS = 500;
|
||||
const SEG_RE = /^seg_(\d+)\.ts$/;
|
||||
/** @type {Map<string,{proc:import('node:child_process').ChildProcess|null,dir:string,upstreamUrl:string,lastAccess:number,nextSeg:number,respawnTimer:NodeJS.Timeout|null,stopping:boolean}>} */
|
||||
const active = new Map();
|
||||
let available = null;
|
||||
const isAvailable = () => {
|
||||
if (available !== null) return available;
|
||||
try {
|
||||
const probe = spawn('ffmpeg', ['-version'], { stdio: 'ignore' });
|
||||
probe.on('error', () => { available = false; });
|
||||
available = true;
|
||||
} catch {
|
||||
available = false;
|
||||
}
|
||||
return available;
|
||||
};
|
||||
/** Sorted ascending list of segment numbers currently on disk */
|
||||
const listSegments = async (dir) => {
|
||||
let names = [];
|
||||
try { names = await fsp.readdir(dir); } catch { return []; }
|
||||
return names
|
||||
.map((n) => SEG_RE.exec(n))
|
||||
.filter(Boolean)
|
||||
.map((m) => Number(m[1]))
|
||||
.sort((a, b) => a - b);
|
||||
};
|
||||
/** seg number -> duration, from every list_*.csv in the dir. */
|
||||
const readDurations = async (dir) => {
|
||||
const out = new Map();
|
||||
let names = [];
|
||||
try { names = await fsp.readdir(dir); } catch { return out; }
|
||||
for (const n of names) {
|
||||
if (!/^list_\d+\.csv$/.test(n)) continue;
|
||||
const txt = await fsp.readFile(path.join(dir, n), 'utf8').catch(() => '');
|
||||
for (const line of txt.split('\n')) {
|
||||
const [file, start, end] = line.split(',');
|
||||
const m = file && SEG_RE.exec(file.trim());
|
||||
if (m && start && end) out.set(Number(m[1]), Math.max(0.1, Number(end) - Number(start)));
|
||||
entry.chunklistUrl = null;
|
||||
entry.failures++;
|
||||
if (entry.failures >= 3) { stop(entry.cameraId, entry.token); return; }
|
||||
} finally {
|
||||
if (!entry.stopping) {
|
||||
entry.timer = setTimeout(() => { entry.polling = poll(entry); }, limits.pollMs);
|
||||
entry.timer.unref?.();
|
||||
}
|
||||
}
|
||||
return out;
|
||||
};
|
||||
const spawnProducer = (cameraId, entry) => {
|
||||
if (entry.stopping) return;
|
||||
entry.boundaries.push(entry.nextSeg);
|
||||
const proc = spawn('ffmpeg', [
|
||||
'-nostdin', '-loglevel', 'warning',
|
||||
...(/^rtmp/i.test(entry.upstreamUrl)
|
||||
? ['-rtmp_live', 'live', '-rw_timeout', '10000000',
|
||||
'-fflags', 'nobuffer', '-analyzeduration', '1000000', '-probesize', '500000']
|
||||
: ['-user_agent', 'gods-eye-view-cctv-proxy/1.0',
|
||||
'-reconnect', '1', '-reconnect_on_network_error', '1', '-reconnect_delay_max', '1']),
|
||||
'-i', entry.upstreamUrl,
|
||||
'-c', 'copy',
|
||||
'-f', 'segment',
|
||||
'-segment_format', 'mpegts',
|
||||
'-segment_time', String(SEGMENT_SECONDS),
|
||||
'-segment_start_number', String(entry.nextSeg),
|
||||
'-segment_list_type', 'csv',
|
||||
'-segment_list_size', '0',
|
||||
'-segment_list', path.join(entry.dir, `list_${entry.nextSeg}.csv`),
|
||||
'-reset_timestamps', '0',
|
||||
path.join(entry.dir, 'seg_%d.ts'),
|
||||
], { stdio: ['ignore', 'ignore', 'pipe'] });
|
||||
entry.proc = proc;
|
||||
proc.stderr?.on('data', (d) => {
|
||||
const line = String(d).trim();
|
||||
if (line) console.warn(`[CCTV:ffmpeg ${cameraId}]`, line);
|
||||
});
|
||||
proc.on('exit', async (code) => {
|
||||
if (entry.proc !== proc) return;
|
||||
entry.proc = null;
|
||||
const segs = await listSegments(entry.dir);
|
||||
entry.nextSeg = segs.length ? segs[segs.length - 1] + 1 : entry.nextSeg;
|
||||
if (entry.stopping || active.get(cameraId) !== entry) return;
|
||||
const idle = Date.now() - entry.lastAccess > IDLE_MS;
|
||||
console.warn(`[CCTV:ffmpeg ${cameraId}] exited (${code}); next seg ${entry.nextSeg}${idle ? ' - idle, not respawning' : ' - respawning'}`);
|
||||
if (idle) return;
|
||||
entry.respawnTimer = setTimeout(() => {
|
||||
entry.respawnTimer = null;
|
||||
spawnProducer(cameraId, entry);
|
||||
}, RESPAWN_DELAY_MS);
|
||||
});
|
||||
};
|
||||
const ensure = async (cameraId, upstreamUrl) => {
|
||||
const ensure = async (cameraId, url, leaseId = 'legacy') => {
|
||||
if (closed) throw new Error('HLS service closed');
|
||||
sameOriginHlsUrl(url, url);
|
||||
let entry = active.get(cameraId);
|
||||
if (entry) {
|
||||
entry.lastAccess = Date.now();
|
||||
if (!entry.proc && !entry.respawnTimer && !entry.stopping) spawnProducer(cameraId, entry);
|
||||
return entry;
|
||||
}
|
||||
const dir = path.join(OUT_ROOT, cameraId.replace(/[^a-zA-Z0-9_-]/g, '_'));
|
||||
await fsp.mkdir(dir, { recursive: true });
|
||||
const existing = await listSegments(dir);
|
||||
entry = {
|
||||
proc: null,
|
||||
dir,
|
||||
upstreamUrl,
|
||||
lastAccess: Date.now(),
|
||||
nextSeg: existing.length ? existing[existing.length - 1] + 1 : 0,
|
||||
respawnTimer: null,
|
||||
stopping: false,
|
||||
boundaries: [],
|
||||
};
|
||||
if (entry && entry.url !== url) { stop(cameraId); entry = null; }
|
||||
if (entry) { touch(entry, leaseId, true); return entry; }
|
||||
// Reserve synchronously, before any await: concurrent creation is singleflight.
|
||||
if (active.size >= limits.sessions) throw new Error('HLS session capacity reached');
|
||||
entry = { cameraId, url, token: randomUUID(), controller: new AbortController(), segments: new Map(), bytes: 0, stopping: false, failures: 0, nextSeq: 0, upstream: new Map(), leases: new Map(), upstreamNewest: -1, pendingDiscontinuity: false, discontinuitiesRemoved: 0 };
|
||||
active.set(cameraId, entry);
|
||||
spawnProducer(cameraId, entry);
|
||||
touch(entry, leaseId, true);
|
||||
entry.polling = poll(entry);
|
||||
return entry;
|
||||
};
|
||||
const waitReady = async (entry) => {
|
||||
const deadline = Date.now() + READY_TIMEOUT_MS;
|
||||
while (Date.now() < deadline) {
|
||||
const segs = await listSegments(entry.dir);
|
||||
if (segs.length >= MIN_READY_SEGMENTS) return true;
|
||||
await new Promise((r) => setTimeout(r, 250));
|
||||
const waitReady = async (entry, signal) => {
|
||||
const until = Date.now() + limits.readyMs;
|
||||
while (!entry.stopping && !signal?.aborted && Date.now() < until) {
|
||||
if (entry.segments.size >= 2) return true;
|
||||
await new Promise((resolve) => setTimeout(resolve, 50));
|
||||
}
|
||||
return false;
|
||||
};
|
||||
/**
|
||||
* Build the playlist from the segments on disk and prune old files.
|
||||
* Newest segment may still be mid-write so it is excluded.
|
||||
* @returns {Promise<string|null>} playlist text, or null if not enough segments.
|
||||
*/
|
||||
const buildPlaylist = async (entry, cameraId) => {
|
||||
const segs = await listSegments(entry.dir);
|
||||
if (segs.length < 2) return null;
|
||||
const complete = segs.slice(0, -1);
|
||||
const window = complete.slice(-PLAYLIST_WINDOW);
|
||||
const keepFrom = segs.length > KEEP_ON_DISK ? segs[segs.length - KEEP_ON_DISK] : -1;
|
||||
for (const n of segs) {
|
||||
if (n < keepFrom) fsp.rm(path.join(entry.dir, `seg_${n}.ts`), { force: true }).catch(() => {});
|
||||
}
|
||||
const prefix = `/api/cctv/media/${encodeURIComponent(cameraId)}/`;
|
||||
const first = window[0];
|
||||
const priorBoundaries = entry.boundaries.filter((b) => b > 0 && b <= first).length;
|
||||
const durations = await readDurations(entry.dir);
|
||||
const maxDur = Math.max(SEGMENT_SECONDS, ...window.map((n) => durations.get(n) || SEGMENT_SECONDS));
|
||||
const lines = [
|
||||
'#EXTM3U',
|
||||
'#EXT-X-VERSION:3',
|
||||
`#EXT-X-TARGETDURATION:${Math.ceil(maxDur)}`,
|
||||
`#EXT-X-MEDIA-SEQUENCE:${first}`,
|
||||
`#EXT-X-DISCONTINUITY-SEQUENCE:${priorBoundaries}`,
|
||||
];
|
||||
let prev = null;
|
||||
for (const n of window) {
|
||||
const isBoundary = n !== first && entry.boundaries.includes(n);
|
||||
const isGap = prev !== null && n !== prev + 1;
|
||||
const dur = durations.get(n) || SEGMENT_SECONDS;
|
||||
if (isBoundary || isGap) lines.push('#EXT-X-DISCONTINUITY');
|
||||
lines.push(`#EXTINF:${dur.toFixed(3)},`, `${prefix}seg_${n}.ts`);
|
||||
prev = n;
|
||||
const buildPlaylist = async (entry, cameraId, leaseId = 'legacy') => {
|
||||
if (entry.stopping || entry.segments.size < 2) return null;
|
||||
touch(entry, leaseId);
|
||||
const segments = [...entry.segments.values()].sort((a, b) => a.seq - b.seq);
|
||||
const lines = ['#EXTM3U', '#EXT-X-VERSION:3', `#EXT-X-TARGETDURATION:${Math.ceil(Math.max(...segments.map((s) => s.duration)))}`, `#EXT-X-MEDIA-SEQUENCE:${segments[0].seq}`, `#EXT-X-DISCONTINUITY-SEQUENCE:${entry.discontinuitiesRemoved}`];
|
||||
let previous;
|
||||
for (const segment of segments) {
|
||||
if (segment.discontinuity || (previous !== undefined && segment.seq !== previous + 1)) lines.push('#EXT-X-DISCONTINUITY');
|
||||
lines.push(`#EXTINF:${segment.duration.toFixed(3)},`, `/api/cctv/media/${encodeURIComponent(cameraId)}/seg_${segment.seq}.ts?session=${entry.token}&lease=${encodeURIComponent(leaseId)}`);
|
||||
previous = segment.seq;
|
||||
}
|
||||
return lines.join('\n') + '\n';
|
||||
};
|
||||
const stopEntry = (entry) => {
|
||||
entry.stopping = true;
|
||||
if (entry.respawnTimer) { clearTimeout(entry.respawnTimer); entry.respawnTimer = null; }
|
||||
if (entry.proc) entry.proc.kill('SIGTERM');
|
||||
fsp.rm(entry.dir, { recursive: true, force: true }).catch(() => {});
|
||||
const getSegment = (cameraId, token, seq, leaseId = 'legacy') => {
|
||||
const entry = active.get(cameraId);
|
||||
if (!entry || entry.token !== token || !entry.leases.has(leaseId)) return null;
|
||||
const body = entry.segments.get(seq)?.body;
|
||||
if (body) touch(entry, leaseId);
|
||||
return body || null;
|
||||
};
|
||||
const sweep = setInterval(async () => {
|
||||
const now = Date.now();
|
||||
for (const [id, entry] of active) {
|
||||
if (now - entry.lastAccess > IDLE_MS) { stopEntry(entry); active.delete(id); continue; }
|
||||
if (entry.proc) {
|
||||
const segs = await listSegments(entry.dir);
|
||||
const newest = segs.length ? segs[segs.length - 1] : -1;
|
||||
if (newest !== entry.watchSeg) { entry.watchSeg = newest; entry.watchAt = now; }
|
||||
else if (now - (entry.watchAt || now) > STALL_MS) {
|
||||
console.warn(`[CCTV:ffmpeg ${id}] no new segment for ${STALL_MS / 1000}s — killing producer`);
|
||||
entry.watchAt = now;
|
||||
entry.proc.kill('SIGKILL');
|
||||
}
|
||||
}
|
||||
}
|
||||
}, 5_000);
|
||||
sweep.unref?.();
|
||||
const shutdown = () => {
|
||||
clearInterval(sweep);
|
||||
for (const [, entry] of active) stopEntry(entry);
|
||||
active.clear();
|
||||
const shutdown = async () => {
|
||||
closed = true;
|
||||
const entries = [...active.values()];
|
||||
for (const entry of entries) stop(entry.cameraId);
|
||||
await Promise.allSettled(entries.map((entry) => entry.polling));
|
||||
};
|
||||
|
||||
return { isAvailable, ensure, waitReady, buildPlaylist, shutdown };
|
||||
}
|
||||
return { ensure, waitReady, buildPlaylist, getSegment, stop, release, shutdown,
|
||||
stats: () => ({ sessions: active.size, bytes: [...active.values()].reduce((total, e) => total + e.bytes, 0) }) };
|
||||
}
|
||||
Reference in new issue
Block a user