ws-pipe-relay

app/public/index.html

@@ -17,22 +17,21 @@
     const player = mpegts.createPlayer(
       { type: 'mpegts', isLive: true, url: wsUrl, hasAudio: false },
       {
-        enableWorker: true,              // transmuxing в отдельном треде
+        enableWorker: true,             // transmuxing в отдельном треде
-        enableStashBuffer: false,        // без IO-stash для real-time
+        enableStashBuffer: true,        // включить stash
-        autoCleanupSourceBuffer: true,   // не копить старые буферы в MSE
+        stashInitialSize: 384,
+
+        autoCleanupSourceBuffer: true,  // не копить старые буферы в MSE
         autoCleanupMaxBackwardDuration: 10,
         autoCleanupMinBackwardDuration: 5,
 
         // liveSync — плавное догоняние через playbackRate, а не жёсткий seek
         liveSync: true,
-        liveSyncMaxLatency: 3.0,         // при плохом интернете даём 3с
+        liveSyncTargetLatency: 4.0,    // target после выравнивания ← дать буферу вырасти
-        liveSyncTargetLatency: 1.5,      // target после выравнивания
+        liveSyncMaxLatency: 10.0,      // при плохом интернете даём 10с ← не трогать буфер до 10s
-        liveSyncPlaybackRate: 1.2,       // ускорение не более 1.2x
+        liveSyncPlaybackRate: 1.05,    // ускорение не более 1.2x ← очень мягкое догоняние
 
-        // liveBufferLatencyChasing оставляем, но увеличиваем окно
+        liveBufferLatencyChasing: false,  // включить stash ← убрать конфликтующий механизм
-        liveBufferLatencyChasing: true,
-        liveBufferLatencyMaxLatency: 3.0,
-        liveBufferLatencyMinRemain: 0.5,
 
         lazyLoad: false,
         deferLoadAfterSourceOpen: false,

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'
 

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<string, WebSocket>()
+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)
+  }
+}
+
+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 })
+})

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}
+      WS_PIPE_PORT: 9091 # WebSocket port for incoming and outgoing binary data
-      CLOUD_INGEST_API_KEY: ${CLOUD_INGEST_API_KEY:-}
+      CLOUD_INGEST_URL: ${CLOUD_INGEST_URL}
-      DEMO_CLOUD_INGEST_URL: ${DEMO_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