diff --git a/app/public/index.html b/app/public/index.html index a7c85bb..0b957e0 100644 --- a/app/public/index.html +++ b/app/public/index.html @@ -17,22 +17,21 @@ const player = mpegts.createPlayer( { type: 'mpegts', isLive: true, url: wsUrl, hasAudio: false }, { - enableWorker: true, // transmuxing в отдельном треде - enableStashBuffer: false, // без IO-stash для real-time - autoCleanupSourceBuffer: true, // не копить старые буферы в MSE + enableWorker: true, // transmuxing в отдельном треде + enableStashBuffer: true, // включить stash + stashInitialSize: 384, + + autoCleanupSourceBuffer: true, // не копить старые буферы в MSE autoCleanupMaxBackwardDuration: 10, autoCleanupMinBackwardDuration: 5, // liveSync — плавное догоняние через playbackRate, а не жёсткий seek liveSync: true, - liveSyncMaxLatency: 3.0, // при плохом интернете даём 3с - liveSyncTargetLatency: 1.5, // target после выравнивания - liveSyncPlaybackRate: 1.2, // ускорение не более 1.2x + liveSyncTargetLatency: 4.0, // target после выравнивания ← дать буферу вырасти + liveSyncMaxLatency: 10.0, // при плохом интернете даём 10с ← не трогать буфер до 10s + liveSyncPlaybackRate: 1.05, // ускорение не более 1.2x ← очень мягкое догоняние - // liveBufferLatencyChasing оставляем, но увеличиваем окно - liveBufferLatencyChasing: true, - liveBufferLatencyMaxLatency: 3.0, - liveBufferLatencyMinRemain: 0.5, + liveBufferLatencyChasing: false, // включить stash ← убрать конфликтующий механизм lazyLoad: false, deferLoadAfterSourceOpen: false, diff --git a/app/src/index.ts b/app/src/index.ts index 9fbd394..f70bf23 100644 --- a/app/src/index.ts +++ b/app/src/index.ts @@ -2,6 +2,7 @@ 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 new file mode 100644 index 0000000..ef0ab7d --- /dev/null +++ b/app/src/stream/ws-pipe-relay.ts @@ -0,0 +1,94 @@ +import type { IncomingMessage } 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 LOG_INTERVAL_MS = 10_000 + +const wss = new WebSocketServer({ port: wsPipePort, perMessageDeflate: false }) + +const producersByChannel = new Map() +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) + } +} + +function handleProducer(ws: WebSocket, channel: string): void { + const existing = producersByChannel.get(channel) + if (existing?.readyState === WebSocket.OPEN) { + existing.close(1000, 'replaced') + } + producersByChannel.set(channel, ws) + + ws.on('message', (data, isBinary) => { + if (!isBinary) return + const chunk = Buffer.isBuffer(data) ? data : Buffer.from(data as ArrayBuffer) + logThroughput(channel, chunk.length) + broadcast(channel, chunk) + }) + + ws.on('close', () => { + if (producersByChannel.get(channel) === ws) { + producersByChannel.delete(channel) + } + }) +} + +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] ?? '' + + if (path.startsWith('/in/')) { + handleProducer(ws, lastTopicSegment(path)) + return + } + + 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 f96b011..4cec20e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -17,13 +17,15 @@ services: expose: - 80 - 9090 + - 9091 environment: MQTT_URL: mqtt://mosquitto:1883 HTTP_PORT: 80 # http port for video previewiong page WS_PORT: 9090 # WebSocket port for video streaming - CLOUD_INGEST_URL: ${CLOUD_INGEST_URL:-https://demo.backend.drill.greact.ru/ingest} - CLOUD_INGEST_API_KEY: ${CLOUD_INGEST_API_KEY:-} - DEMO_CLOUD_INGEST_URL: ${DEMO_CLOUD_INGEST_URL:-} + WS_PIPE_PORT: 9091 # WebSocket port for incoming and outgoing binary data + CLOUD_INGEST_URL: ${CLOUD_INGEST_URL} + CLOUD_INGEST_API_KEY: ${CLOUD_INGEST_API_KEY} + DEMO_CLOUD_INGEST_URL: ${DEMO_CLOUD_INGEST_URL} EDGE5_MODBUS_V2_MAPPING_FILE: ${EDGE5_MODBUS_V2_MAPPING_FILE:-/app/src/mappings/edge5-modbus.json} networks: - default