diff --git a/app/VIDEO.md b/app/VIDEO.md new file mode 100644 index 0000000..cd6041d --- /dev/null +++ b/app/VIDEO.md @@ -0,0 +1,96 @@ +Финальные требования и настройки у нас сейчас такие. + +## Финальный пресет чанкования ffmpeg (RTSP → MPEG‑TS → stdout) + +Мы остановились на сценарии, где ffmpeg: + +- тянет RTSP **по TCP** с камер, +- минимизирует внутреннюю буферизацию и время старта, +- не перекодирует видео (копирует H.264 как есть); если возникнут проблемы с нагрузкой на сеть, можно поискать варианты снизить; но на данном этапе важнее "накормить" видеобуфер воспроизведения хоть чем-нибудь; +- выдаёт результат в **stdout (`pipe:1`)**, откуда его забирает mqtt транспорт. + +Финальный набор аргументов: + +```js +[ + '-hide_banner', + '-fflags', 'nobuffer', // не буферировать при probe stream info + '-flags', 'low_delay', // включить low-delay на уровне декодера + '-analyzeduration', '100000', // ограничить анализ потока 100 ms (уменьшить стартовую задержку) + '-probesize', '32768', // зондировать только 32 KB (быстрее старт, меньше буфера) + '-flush_packets', '1', // сразу сбрасывать пакеты, не копить их внутри muxer'а + + '-rtsp_transport', 'tcp', // тянуть RTSP по TCP (надёжнее на плохой сети) + '-i', 'rtsp://admin:admin@192.168.0.11:554/live/sub', // sub-поток камеры + + '-an', // отключить аудио — нужен только видеофрейм + '-c:v', 'copy', // не перекодировать видео, просто копировать кадры + '-f', 'mpegts', // контейнер MPEG-TS для дальнейшего трансмультиплексирования + '-mpegts_flags', 'resend_headers', // периодически повторять заголовки для надёжного декода + 'pipe:1' // выводить поток в stdout (дальше забирает твой транспорт) +] +``` + +Ключевые моменты: + +- `-c:v copy` сохраняет оригинальный кадр и GOP камеры — минимум задержки и вычислительной нагрузки. +- `-fflags nobuffer + -flags low_delay + -flush_packets 1` минимизируют буферизацию и увеличивают «прозрачность» для твоего транспорта. +- `-rtsp_transport tcp` это про **вход** (камера → ffmpeg), а `pipe:1` — про **выход** (ffmpeg → mqtt). + +*** + +## Финальный пресет воспроизведения (mpegts.js + HTML5 video) + +На клиенте мы на данном этапе определяемся работать как с «live‑видео с буфером», а не как с чистым real‑time: + +- формат источника: **MPEG‑TS/H.264** по WebSocket; +- плеер: `mpegts.js`, режим `type: 'mpegts', isLive: true`; +- буфер: небольшой stash + несколько секунд latency для сглаживания провалов. + +Финальные параметры инициализации: + +```js +const wsUrl = 'wss://...'; + +const player = mpegts.createPlayer( + { + type: 'mpegts', // MPEG-TS поток по WS + isLive: true, + url: wsUrl, + hasAudio: false, // видео без аудио + }, + { + enableWorker: true, + + // Включаем stash, чтобы сглаживать редкий / рваный поток + enableStashBuffer: true, + stashInitialSize: 512, // немного увеличенный stash для накопления + + autoCleanupSourceBuffer: true, + autoCleanupMaxBackwardDuration: 10, // держим минимум истории + autoCleanupMinBackwardDuration: 5, + + // Разрешаем latency-chasing с «комфортной» задержкой + liveBufferLatencyChasing: true, + liveBufferLatencyMaxLatency: 5.0, // целевая задержка до ~5 секунд + liveBufferLatencyMinRemain: 2.0, // оставляем ~2 секунды буфера после сброса + + liveSync: false, // не пытаемся мягко догонять — приоритет стабильности + lazyLoad: false, + deferLoadAfterSourceOpen: false, + } +); + +player.attachMediaElement(document.getElementById('v')); +player.load(); +player.play(); +``` + +Смысл этих настроек: + +- **Stash‑buffer включён**: mpegts.js может аккумулировать некоторое количество данных, прежде чем кормить MSE (браузеру), что хорошо при «дырявом» канале. +- **Latency‑chasing с MaxLatency ≈ 5 s**: браузер допускает несколько секунд отставания, но не позволяет задержке расти бесконечно; при накоплении лишнего буфера он «подпрыгивает» ближе к live‑позиции, сохраняя `MinRemain` ~2 секунды запаса. +- `liveSync: false` — мы не гонимся за минимально возможной задержкой любой ценой, а принимаем небольшой «запас» как разумный компромисс. + +При улучшении канала или выполнениии работ по снижению нагрузки на сеть (уменьшения битрейта, fps) +нужно грамотно управлять буфером на стороне клиента (stash + latency). diff --git a/app/public/index.html b/app/public/index.html index 910f0d5..caa49e5 100644 --- a/app/public/index.html +++ b/app/public/index.html @@ -16,6 +16,26 @@ const wsUrl = `${location.protocol === 'https:' ? 'wss' : 'ws'}://${location.host}/ws/${cameraId}`; const player = mpegts.createPlayer( { type: 'mpegts', isLive: true, url: wsUrl, hasAudio: false }, + { + enableWorker: true, + + // Включаем stash, чтобы сглаживать редкий / рваный поток + enableStashBuffer: true, + stashInitialSize: 512, // немного увеличенный stash для накопления + + autoCleanupSourceBuffer: true, + autoCleanupMaxBackwardDuration: 10, // держим минимум истории + autoCleanupMinBackwardDuration: 5, + + // Разрешаем latency-chasing с «комфортной» задержкой + liveBufferLatencyChasing: true, + liveBufferLatencyMaxLatency: 5.0, // целевая задержка до ~5 секунд + liveBufferLatencyMinRemain: 2.0, // оставляем ~2 секунды буфера после сброса + + liveSync: false, // не пытаемся мягко догонять — приоритет стабильности + lazyLoad: false, + deferLoadAfterSourceOpen: false, + } ); player.attachMediaElement(document.getElementById('v')) player.load() diff --git a/app/src/index.ts b/app/src/index.ts index f70bf23..9fbd394 100644 --- a/app/src/index.ts +++ b/app/src/index.ts @@ -2,7 +2,6 @@ import './handlers/index.js' import './http/server.js' import './mqtt.js' import './stream/edge-chunk-relay.js' -import './stream/ws-pipe-relay.js' import { log } from './helpers/log.js' diff --git a/app/src/stream/ws-pipe-relay.ts b/app/src/stream/ws-pipe-relay.ts deleted file mode 100644 index 4438a63..0000000 --- a/app/src/stream/ws-pipe-relay.ts +++ /dev/null @@ -1,102 +0,0 @@ -import type { IncomingMessage } from 'http' -import { createServer } from 'http' -import { WebSocket, WebSocketServer } from 'ws' - -import { log } from '../helpers/log.js' -import { lastTopicSegment } from '../helpers/topic-match.js' - -const wsPipePort = Number(process.env.WS_PIPE_PORT) -const httpIngestPort = Number(process.env.HTTP_INGEST_PORT) // новый порт, например 3456 -const LOG_INTERVAL_MS = 10_000 - -const wss = new WebSocketServer({ port: wsPipePort, perMessageDeflate: false }) - -const viewersByChannel = new Map>() -const lastLogByChannel = new Map() - -function getViewers(channel: string): Set { - let viewers = viewersByChannel.get(channel) - if (!viewers) { - viewers = new Set() - viewersByChannel.set(channel, viewers) - } - return viewers -} - -function broadcast(channel: string, chunk: Buffer): void { - const clients = viewersByChannel.get(channel) - if (!clients) return - for (const client of clients) { - if (client.readyState === WebSocket.OPEN) { - client.send(chunk) - } - } -} - -function logThroughput(channel: string, bytes: number): void { - const now = Date.now() - const lastLog = lastLogByChannel.get(channel) ?? 0 - if (now - lastLog >= LOG_INTERVAL_MS) { - log(`pipe.in [${channel}] ↑ ${bytes}B`) - lastLogByChannel.set(channel, now) - } -} - -// --- HTTP ingest --- -const httpIngest = createServer((req: IncomingMessage, res) => { - const path = req.url?.split('?')[0] ?? '' - - if (!path.startsWith('/in/')) { - res.writeHead(404).end() - return - } - - const channel = lastTopicSegment(path) - log(`pipe.in.http_connected [${channel}]`) - - req.on('data', (chunk: Buffer) => { - logThroughput(channel, chunk.length) - broadcast(channel, chunk) - }) - - req.on('end', () => { - log(`pipe.in.http_disconnected [${channel}]`) - res.writeHead(200).end() - }) - - req.on('error', (err) => { - log(`pipe.in.http_error [${channel}]`, err) - res.writeHead(500).end() - }) -}) - -httpIngest.listen(httpIngestPort, () => { - log('pipe.http_ingest_started', { port: httpIngestPort }) -}) - -function handleViewer(ws: WebSocket, channel: string, path: string): void { - log(`pipe.out.connection ${path}`) - const viewers = getViewers(channel) - viewers.add(ws) - - ws.on('close', () => { - viewers.delete(ws) - if (viewers.size === 0) viewersByChannel.delete(channel) - }) -} - -wss.on('connection', (ws: WebSocket, req: IncomingMessage) => { - const path = req.url?.split('?')[0] ?? '' - req.socket.setNoDelay(true); - - if (path.startsWith('/out/')) { - handleViewer(ws, lastTopicSegment(path), path) - return - } - - ws.close(1008, 'unknown path') -}) - -wss.on('listening', () => { - log('pipe.server_started', { port: wsPipePort }) -}) diff --git a/docker-compose.yml b/docker-compose.yml index 5f4b6bd..f2a8b3c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -17,14 +17,10 @@ services: expose: - 80 - 9090 - - 9091 - - 9092 environment: MQTT_URL: mqtt://mosquitto:1883 - HTTP_PORT: 80 # http port for video previewiong page + HTTP_PORT: 80 # http port for video previewing page WS_PORT: 9090 # WebSocket port for video streaming - WS_PIPE_PORT: 9091 # WebSocket port for outgoing binary data - HTTP_INGEST_PORT: 9092 # HTTP port for incoming binary data CLOUD_INGEST_URL: ${CLOUD_INGEST_URL} CLOUD_INGEST_API_KEY: ${CLOUD_INGEST_API_KEY} DEMO_CLOUD_INGEST_URL: ${DEMO_CLOUD_INGEST_URL}