542 lines
16 KiB
TypeScript
542 lines
16 KiB
TypeScript
import { HttpStatusCode } from '@peertube/peertube-models'
|
|
import { logger } from '@server/helpers/logger.js'
|
|
import { buildRtspSinkUrlForTargetCamera } from '@server/lib/live-videos/rtsp-url.js'
|
|
import { LiveVideoModel, LiveVideoStatus } from '@server/models/video/live-video.js'
|
|
import { CONFIG } from '@server/initializers/config.js'
|
|
import { asyncMiddleware } from '@server/middlewares/async.js'
|
|
import express from 'express'
|
|
import { readFile } from 'node:fs/promises'
|
|
import { existsSync } from 'node:fs'
|
|
import { mkdir, rm } from 'node:fs/promises'
|
|
import { join, resolve } from 'node:path'
|
|
import { setTimeout as wait } from 'node:timers/promises'
|
|
import { ChildProcessWithoutNullStreams, spawn } from 'node:child_process'
|
|
import yaml from 'js-yaml'
|
|
|
|
type StreamConfig = {
|
|
streams?: Record<string, string | string[]>
|
|
}
|
|
|
|
type LiveSession = {
|
|
ffmpegProcess?: ChildProcessWithoutNullStreams
|
|
keepAliveUntil?: number
|
|
startedAt: number
|
|
lastAccessAt: number
|
|
startingPromise?: Promise<void>
|
|
restartTimer?: NodeJS.Timeout
|
|
lastError?: string
|
|
stopped?: boolean
|
|
}
|
|
|
|
const SESSION_IDLE_TIMEOUT_MS = 300_000
|
|
const SESSION_SWEEP_INTERVAL_MS = 20_000
|
|
const MANIFEST_WAIT_TIMEOUT_MS = 20_000
|
|
const MANIFEST_WAIT_INTERVAL_MS = 250
|
|
const LIVE_CONFIG_PATH = join(process.cwd(), 'livestream-sources.yaml')
|
|
const LIVE_WORKDIR = join(CONFIG.STORAGE.TMP_PERSISTENT_DIR, 'native-live')
|
|
|
|
const liveStreamRouter = express.Router()
|
|
const liveSessions = new Map<string, LiveSession>()
|
|
|
|
if(!CONFIG.LIVE.AUTO_START_FFMPEG_STREAMING) {
|
|
const sweepTimer = setInterval(cleanupInactiveSessions, SESSION_SWEEP_INTERVAL_MS)
|
|
sweepTimer.unref()
|
|
}
|
|
|
|
liveStreamRouter.get('/streams', asyncMiddleware(listStreams))
|
|
liveStreamRouter.get('/streams/:streamKey/index.m3u8', asyncMiddleware(getManifest))
|
|
liveStreamRouter.get('/streams/:streamKey/:segmentName', asyncMiddleware(getSegment))
|
|
liveStreamRouter.get('/scheduled-streams/:cameraId/index.m3u8', asyncMiddleware(getScheduledManifest))
|
|
liveStreamRouter.get('/scheduled-streams/:cameraId/:segmentName', asyncMiddleware(getScheduledSegment))
|
|
|
|
export { liveStreamRouter, initLiveStreams, startScheduledLiveSession, stopScheduledLiveSession }
|
|
|
|
async function listStreams (req: express.Request, res: express.Response) {
|
|
const streams = await readConfiguredStreams()
|
|
|
|
return res.json({
|
|
data: Object.keys(streams)
|
|
})
|
|
}
|
|
|
|
async function getManifest (req: express.Request, res: express.Response) {
|
|
const streamKey = sanitizeStreamKey(req.params.streamKey)
|
|
|
|
if (!streamKey) {
|
|
return res.fail({
|
|
status: HttpStatusCode.BAD_REQUEST_400,
|
|
message: 'Invalid stream key.'
|
|
})
|
|
}
|
|
|
|
const sourceUrl = await findStreamSourceUrl(streamKey)
|
|
if (!sourceUrl) {
|
|
return res.fail({
|
|
status: HttpStatusCode.NOT_FOUND_404,
|
|
message: `Unknown stream key "${streamKey}".`
|
|
})
|
|
}
|
|
|
|
return serveManifest(res, streamKey, sourceUrl)
|
|
}
|
|
|
|
async function getScheduledManifest (req: express.Request, res: express.Response) {
|
|
const cameraId = sanitizeStreamKey(req.params.cameraId)
|
|
|
|
if (!cameraId) {
|
|
return res.fail({
|
|
status: HttpStatusCode.BAD_REQUEST_400,
|
|
message: 'Invalid camera id.'
|
|
})
|
|
}
|
|
|
|
const activeLiveVideo = await loadActiveScheduledLiveVideo(cameraId)
|
|
if (!activeLiveVideo) {
|
|
await stopScheduledLiveSession(cameraId)
|
|
.catch(err => logger.error('Cannot stop stale scheduled sink stream session after expired manifest request.', { err, cameraId }))
|
|
|
|
return res.fail({
|
|
status: HttpStatusCode.NOT_FOUND_404,
|
|
message: 'No live stream is currently active for this camera.'
|
|
})
|
|
}
|
|
|
|
return serveManifest(res, buildScheduledStreamSessionKey(cameraId), buildRtspSinkUrlForTargetCamera(cameraId))
|
|
}
|
|
|
|
async function getSegment (req: express.Request, res: express.Response) {
|
|
const streamKey = sanitizeStreamKey(req.params.streamKey)
|
|
const segmentName = req.params.segmentName
|
|
|
|
if (!streamKey) {
|
|
return res.fail({
|
|
status: HttpStatusCode.BAD_REQUEST_400,
|
|
message: 'Invalid stream key.'
|
|
})
|
|
}
|
|
|
|
if (!segmentName || !/^segment_[0-9]{6}\.ts$/.test(segmentName)) {
|
|
return res.fail({
|
|
status: HttpStatusCode.BAD_REQUEST_400,
|
|
message: 'Invalid segment name.'
|
|
})
|
|
}
|
|
|
|
const sourceUrl = await findStreamSourceUrl(streamKey)
|
|
if (!sourceUrl) {
|
|
return res.fail({
|
|
status: HttpStatusCode.NOT_FOUND_404,
|
|
message: `Unknown stream key "${streamKey}".`
|
|
})
|
|
}
|
|
|
|
return serveSegment(res, streamKey, segmentName, sourceUrl)
|
|
}
|
|
|
|
async function getScheduledSegment (req: express.Request, res: express.Response) {
|
|
const cameraId = sanitizeStreamKey(req.params.cameraId)
|
|
const segmentName = req.params.segmentName
|
|
|
|
if (!cameraId) {
|
|
return res.fail({
|
|
status: HttpStatusCode.BAD_REQUEST_400,
|
|
message: 'Invalid camera id.'
|
|
})
|
|
}
|
|
|
|
if (!segmentName || !/^segment_[0-9]{6}\.ts$/.test(segmentName)) {
|
|
return res.fail({
|
|
status: HttpStatusCode.BAD_REQUEST_400,
|
|
message: 'Invalid segment name.'
|
|
})
|
|
}
|
|
|
|
const activeLiveVideo = await loadActiveScheduledLiveVideo(cameraId)
|
|
if (!activeLiveVideo) {
|
|
await stopScheduledLiveSession(cameraId)
|
|
.catch(err => logger.error('Cannot stop stale scheduled sink stream session after expired segment request.', { err, cameraId, segmentName }))
|
|
|
|
return res.fail({
|
|
status: HttpStatusCode.NOT_FOUND_404,
|
|
message: 'No live stream is currently active for this camera.'
|
|
})
|
|
}
|
|
|
|
return serveSegment(res, buildScheduledStreamSessionKey(cameraId), segmentName, buildRtspSinkUrlForTargetCamera(cameraId))
|
|
}
|
|
|
|
async function serveManifest (res: express.Response, streamKey: string, sourceUrl: string) {
|
|
const readyToServe = await ensureLiveSession(streamKey, sourceUrl)
|
|
if (!readyToServe) {
|
|
return res.fail({
|
|
status: HttpStatusCode.NOT_FOUND_404,
|
|
message: 'No live stream is currently active for this camera.'
|
|
})
|
|
}
|
|
|
|
const manifestPath = buildStreamPath(streamKey, 'index.m3u8')
|
|
const ready = await waitForFile(manifestPath, MANIFEST_WAIT_TIMEOUT_MS)
|
|
|
|
if (!ready) {
|
|
const session = liveSessions.get(streamKey)
|
|
|
|
return res.fail({
|
|
status: HttpStatusCode.SERVICE_UNAVAILABLE_503,
|
|
message: session?.lastError || 'Stream is starting, please retry in a few seconds.'
|
|
})
|
|
}
|
|
|
|
touchSession(streamKey)
|
|
res.setHeader('Cache-Control', 'public, max-age=1, must-revalidate')
|
|
return res.sendFile(resolve(manifestPath))
|
|
}
|
|
|
|
async function serveSegment (res: express.Response, streamKey: string, segmentName: string, sourceUrl: string) {
|
|
const readyToServe = await ensureLiveSession(streamKey, sourceUrl)
|
|
if (!readyToServe) {
|
|
return res.fail({
|
|
status: HttpStatusCode.NOT_FOUND_404,
|
|
message: 'No live stream is currently active for this camera.'
|
|
})
|
|
}
|
|
|
|
const segmentPath = buildStreamPath(streamKey, segmentName)
|
|
if (!existsSync(segmentPath)) {
|
|
return res.fail({
|
|
status: HttpStatusCode.NOT_FOUND_404,
|
|
message: 'Segment not found.'
|
|
})
|
|
}
|
|
|
|
touchSession(streamKey)
|
|
res.setHeader('Cache-Control', 'public, max-age=30, immutable')
|
|
return res.sendFile(resolve(segmentPath))
|
|
}
|
|
|
|
async function readConfiguredStreams () {
|
|
if (!existsSync(LIVE_CONFIG_PATH)) {
|
|
return {}
|
|
}
|
|
|
|
const raw = await readFile(LIVE_CONFIG_PATH, 'utf-8')
|
|
const parsed = yaml.load(raw) as StreamConfig | undefined
|
|
return parsed?.streams || {}
|
|
}
|
|
|
|
async function findStreamSourceUrl (streamKey: string) {
|
|
const streams = await readConfiguredStreams()
|
|
const streamValue = streams[streamKey]
|
|
|
|
if (!streamValue) {
|
|
return buildRtspSinkUrlForTargetCamera(streamKey)
|
|
}
|
|
|
|
if (typeof streamValue === 'string') return streamValue
|
|
|
|
const firstRtsp = streamValue.find(v => typeof v === 'string' && v.startsWith('rtsp://'))
|
|
if (firstRtsp) return firstRtsp
|
|
|
|
return streamValue.find(v => typeof v === 'string') || buildRtspSinkUrlForTargetCamera(streamKey)
|
|
}
|
|
|
|
function sanitizeStreamKey (streamKey: string) {
|
|
if (!streamKey) return undefined
|
|
if (!/^[A-Za-z0-9._-]+$/.test(streamKey)) return undefined
|
|
|
|
return streamKey
|
|
}
|
|
|
|
function buildStreamPath (streamKey: string, fileName: string) {
|
|
return join(LIVE_WORKDIR, streamKey, fileName)
|
|
}
|
|
|
|
function buildScheduledStreamSessionKey (cameraId: string) {
|
|
return `scheduled-${cameraId}`
|
|
}
|
|
|
|
async function startScheduledLiveSession (cameraId: string, options: { keepAliveUntil?: Date | number } = {}) {
|
|
const streamKey = buildScheduledStreamSessionKey(cameraId)
|
|
const keepAliveUntil = normalizeKeepAliveUntil(options.keepAliveUntil)
|
|
|
|
return ensureLiveSession(streamKey, buildRtspSinkUrlForTargetCamera(cameraId), { keepAliveUntil })
|
|
}
|
|
|
|
async function stopScheduledLiveSession (cameraId: string) {
|
|
return stopLiveSession(buildScheduledStreamSessionKey(cameraId))
|
|
}
|
|
|
|
function touchSession (streamKey: string) {
|
|
const session = liveSessions.get(streamKey)
|
|
if (!session) return
|
|
|
|
session.lastAccessAt = Date.now()
|
|
}
|
|
|
|
async function ensureLiveSession (streamKey: string, sourceUrl: string, options: { keepAliveUntil?: number } = {}) {
|
|
let session = liveSessions.get(streamKey)
|
|
const now = Date.now()
|
|
|
|
if ((options.keepAliveUntil || 0) <= now && options.keepAliveUntil !== undefined) {
|
|
await stopLiveSession(streamKey)
|
|
return false
|
|
}
|
|
|
|
if (session && (session.keepAliveUntil || 0) <= now && session.keepAliveUntil !== undefined) {
|
|
await stopLiveSession(streamKey)
|
|
return false
|
|
}
|
|
|
|
if (!session) {
|
|
session = {
|
|
startedAt: 0,
|
|
lastAccessAt: Date.now(),
|
|
lastError: undefined,
|
|
stopped: false
|
|
}
|
|
liveSessions.set(streamKey, session)
|
|
}
|
|
|
|
session.stopped = false
|
|
clearRestartTimer(session)
|
|
|
|
if (options.keepAliveUntil) {
|
|
session.keepAliveUntil = Math.max(session.keepAliveUntil || 0, options.keepAliveUntil)
|
|
}
|
|
|
|
if (isChildProcessAlive(session.ffmpegProcess)) {
|
|
touchSession(streamKey)
|
|
return true
|
|
}
|
|
|
|
if (session.startingPromise) {
|
|
await session.startingPromise
|
|
touchSession(streamKey)
|
|
return true
|
|
}
|
|
|
|
session.startingPromise = startSession(streamKey, sourceUrl)
|
|
try {
|
|
await session.startingPromise
|
|
} finally {
|
|
session.startingPromise = undefined
|
|
}
|
|
|
|
touchSession(streamKey)
|
|
return true
|
|
}
|
|
|
|
async function startSession (streamKey: string, sourceUrl: string) {
|
|
const outputDirectory = join(LIVE_WORKDIR, streamKey)
|
|
await rm(outputDirectory, { recursive: true, force: true })
|
|
await mkdir(outputDirectory, { recursive: true })
|
|
|
|
const manifestPath = join(outputDirectory, 'index.m3u8')
|
|
const segmentPattern = join(outputDirectory, 'segment_%06d.ts')
|
|
|
|
const ffmpegPath = CONFIG.LIVE.FFMPEG.PATH?.trim() || 'ffmpeg'
|
|
const segmentDuration = Math.max(1, CONFIG.LIVE.FFMPEG.HLS_TIME)
|
|
const hlsListSize = Math.max(1, CONFIG.LIVE.FFMPEG.HLS_LIST_SIZE)
|
|
|
|
const ffmpegArgs = [
|
|
'-hide_banner',
|
|
'-loglevel', 'warning',
|
|
'-fflags', '+genpts+igndts+discardcorrupt',
|
|
'-rtsp_transport', 'tcp',
|
|
'-thread_queue_size', '512',
|
|
'-i', sourceUrl,
|
|
'-map', '0:v:0',
|
|
'-map', '0:a:0?',
|
|
'-c', 'copy',
|
|
'-f', 'hls',
|
|
'-hls_time', String(segmentDuration),
|
|
'-hls_list_size', String(hlsListSize),
|
|
'-hls_delete_threshold', '2',
|
|
'-hls_flags', 'delete_segments+independent_segments+omit_endlist',
|
|
'-hls_segment_filename', segmentPattern,
|
|
manifestPath
|
|
]
|
|
|
|
const ffmpegProcess = spawn(ffmpegPath, ffmpegArgs, {
|
|
stdio: [ 'ignore', 'ignore', 'pipe' ]
|
|
})
|
|
|
|
const session = liveSessions.get(streamKey)
|
|
if (!session || session.stopped) {
|
|
ffmpegProcess.kill('SIGTERM')
|
|
return
|
|
}
|
|
|
|
session.ffmpegProcess = ffmpegProcess
|
|
session.startedAt = Date.now()
|
|
session.lastAccessAt = Date.now()
|
|
session.lastError = undefined
|
|
|
|
let stderrBuffer = ''
|
|
ffmpegProcess.stderr.on('data', chunk => {
|
|
// Keep pipe drained to avoid stalling ffmpeg and keep last lines for diagnostics.
|
|
const line = chunk?.toString?.() || ''
|
|
stderrBuffer = (stderrBuffer + line).slice(-4_000)
|
|
})
|
|
|
|
ffmpegProcess.once('error', err => {
|
|
logger.error('Cannot start ffmpeg live session', { err, streamKey })
|
|
|
|
const existing = liveSessions.get(streamKey)
|
|
if (!existing) return
|
|
|
|
existing.lastError = `Cannot start ffmpeg: ${err.message}`
|
|
})
|
|
|
|
ffmpegProcess.once('exit', () => {
|
|
const existing = liveSessions.get(streamKey)
|
|
if (!existing) return
|
|
if (existing.ffmpegProcess !== ffmpegProcess) return
|
|
|
|
existing.ffmpegProcess = undefined
|
|
existing.startedAt = 0
|
|
|
|
const trimmedError = stderrBuffer.trim()
|
|
if (trimmedError) {
|
|
const lines = trimmedError.split('\n')
|
|
existing.lastError = lines.slice(-3).join(' | ')
|
|
}
|
|
|
|
const shouldRestart = CONFIG.LIVE.AUTO_START_FFMPEG_STREAMING ||
|
|
((existing.keepAliveUntil || 0) > Date.now())
|
|
|
|
if (shouldRestart && !existing.stopped) {
|
|
const keepAliveUntil = existing.keepAliveUntil
|
|
clearRestartTimer(existing)
|
|
existing.restartTimer = setTimeout(() => {
|
|
const current = liveSessions.get(streamKey)
|
|
if (current !== existing || current?.stopped) return
|
|
|
|
void ensureLiveSession(streamKey, sourceUrl, { keepAliveUntil })
|
|
}, 3_000)
|
|
}
|
|
})
|
|
}
|
|
|
|
async function stopLiveSession (streamKey: string) {
|
|
const session = liveSessions.get(streamKey)
|
|
if (!session) return false
|
|
|
|
session.stopped = true
|
|
session.keepAliveUntil = undefined
|
|
session.startingPromise = undefined
|
|
clearRestartTimer(session)
|
|
|
|
if (session.ffmpegProcess) {
|
|
stopSessionProcess(session.ffmpegProcess, streamKey)
|
|
}
|
|
|
|
liveSessions.delete(streamKey)
|
|
return true
|
|
}
|
|
|
|
async function waitForFile (filePath: string, timeoutMs: number) {
|
|
const end = Date.now() + timeoutMs
|
|
|
|
while (Date.now() < end) {
|
|
if (existsSync(filePath)) return true
|
|
|
|
await wait(MANIFEST_WAIT_INTERVAL_MS)
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
async function initLiveStreams () {
|
|
const streams = await readConfiguredStreams()
|
|
|
|
for (const streamKey of Object.keys(streams)) {
|
|
const sourceUrl = await findStreamSourceUrl(streamKey)
|
|
if (sourceUrl) {
|
|
try {
|
|
await ensureLiveSession(streamKey, sourceUrl)
|
|
} catch (err) {
|
|
logger.warn('Failed to start live stream %s', streamKey, { err })
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
function cleanupInactiveSessions () {
|
|
const now = Date.now()
|
|
|
|
for (const [ streamKey, session ] of liveSessions.entries()) {
|
|
if ((session.keepAliveUntil || 0) > now) {
|
|
continue
|
|
}
|
|
|
|
if (!isChildProcessAlive(session.ffmpegProcess)) {
|
|
liveSessions.delete(streamKey)
|
|
continue
|
|
}
|
|
|
|
if (now - session.lastAccessAt < SESSION_IDLE_TIMEOUT_MS) continue
|
|
stopSessionProcess(session.ffmpegProcess, streamKey)
|
|
liveSessions.delete(streamKey)
|
|
}
|
|
}
|
|
|
|
function normalizeKeepAliveUntil (keepAliveUntil?: Date | number) {
|
|
if (keepAliveUntil instanceof Date) return keepAliveUntil.getTime()
|
|
if (typeof keepAliveUntil === 'number' && Number.isFinite(keepAliveUntil)) return keepAliveUntil
|
|
|
|
return undefined
|
|
}
|
|
|
|
function clearRestartTimer (session: LiveSession) {
|
|
if (!session.restartTimer) return
|
|
|
|
clearTimeout(session.restartTimer)
|
|
session.restartTimer = undefined
|
|
}
|
|
|
|
function isChildProcessAlive (childProcess?: ChildProcessWithoutNullStreams) {
|
|
if (!childProcess?.pid) return false
|
|
if (childProcess.exitCode !== null || childProcess.signalCode !== null) return false
|
|
|
|
try {
|
|
process.kill(childProcess.pid, 0)
|
|
return true
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
|
|
function stopSessionProcess (childProcess: ChildProcessWithoutNullStreams, streamKey: string) {
|
|
const pid = childProcess.pid
|
|
if (!pid || !isChildProcessAlive(childProcess)) return
|
|
|
|
childProcess.kill('SIGTERM')
|
|
|
|
const forceKillTimer = setTimeout(() => {
|
|
if (!isChildProcessAlive(childProcess)) return
|
|
|
|
logger.warn('Force killing stale live session ffmpeg process for %s.', streamKey, { pid })
|
|
|
|
try {
|
|
process.kill(pid, 'SIGKILL')
|
|
} catch (err) {
|
|
logger.warn('Cannot force kill live session ffmpeg process for %s.', streamKey, { err, pid })
|
|
}
|
|
}, 5_000)
|
|
|
|
forceKillTimer.unref()
|
|
}
|
|
|
|
async function loadActiveScheduledLiveVideo (cameraId: string) {
|
|
const now = new Date()
|
|
const liveVideos = await LiveVideoModel.listForStarting({
|
|
startBefore: now,
|
|
endAfter: now
|
|
})
|
|
|
|
return liveVideos.find(liveVideo =>
|
|
liveVideo.targetCamera === cameraId &&
|
|
liveVideo.status === LiveVideoStatus.SCHEDULED
|
|
)
|
|
}
|