finalize video streaming issues
This commit is contained in:
96
app/VIDEO.md
Normal file
96
app/VIDEO.md
Normal file
@@ -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).
|
||||
@@ -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()
|
||||
|
||||
@@ -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'
|
||||
|
||||
|
||||
@@ -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<string, Set<WebSocket>>()
|
||||
const lastLogByChannel = new Map<string, number>()
|
||||
|
||||
function getViewers(channel: string): Set<WebSocket> {
|
||||
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 })
|
||||
})
|
||||
Reference in New Issue
Block a user