ws-pipe-relay

video optimization/
WebSocketServer.clientsByCamera
2026-07-04 21:03:50 +05:00 · 2026-07-04 18:57:41 +05:00 · 2026-07-04 18:09:16 +05:00 · 2026-07-04 17:26:30 +05:00 · 2026-07-03 17:38:23 +05:00 · 2026-07-03 17:30:05 +05:00
15 changed files with 300 additions and 43 deletions
--- a/app/Dockerfile
+++ b/app/Dockerfile
@@ -10,5 +10,9 @@ COPY src ./src

 RUN npm run build && npm prune --omit=dev

+# app/public/index.html has content for video preview
+# TODO move all static files (html, mappers) to a separate folder "assets"
+COPY public ./public 
+
 ENV NODE_ENV=production
 CMD ["node", "dist/index.js"]
--- a/app/public/index.html
+++ b/app/public/index.html
@@ -4,15 +4,39 @@
  <meta charset="utf-8" />
  <title>edge5 video relay</title>
  <script src="https://cdn.jsdelivr.net/npm/mpegts.js/dist/mpegts.js"></script>
+  <style>
+    body { margin: 0; background: #000; }
+    #v { display: block; width: 100vw; height: 100vh; object-fit: contain; }
+  </style>
 </head>
 <body>
-  <video id="v" controls autoplay muted style="width:100%"></video>
+  <video id="v" controls autoplay muted></video>
  <script>
-    const wsUrl = `ws://${location.hostname}:9090`
+    const cameraId = new URLSearchParams(location.search).get('camera') ?? 'v1';
+    const wsUrl = `${location.protocol === 'https:' ? 'wss' : 'ws'}://${location.host}/ws/${cameraId}`;
    const player = mpegts.createPlayer(
-      { type: 'mpegts', isLive: true, url: wsUrl },
-      { liveBufferLatencyChasing: true, liveBufferLatencyMaxLatency: 1.5 }
-    )
+      { type: 'mpegts', isLive: true, url: wsUrl, hasAudio: false },
+      {
+        enableWorker: true,             // transmuxing в отдельном треде
+        enableStashBuffer: true,        // включить stash
+        stashInitialSize: 384,
+
+        autoCleanupSourceBuffer: true,  // не копить старые буферы в MSE
+        autoCleanupMaxBackwardDuration: 10,
+        autoCleanupMinBackwardDuration: 5,
+
+        // liveSync — плавное догоняние через playbackRate, а не жёсткий seek
+        liveSync: true,
+        liveSyncTargetLatency: 4.0,    // target после выравнивания ← дать буферу вырасти
+        liveSyncMaxLatency: 10.0,      // при плохом интернете даём 10с ← не трогать буфер до 10s
+        liveSyncPlaybackRate: 1.05,    // ускорение не более 1.2x ← очень мягкое догоняние
+
+        liveBufferLatencyChasing: false,  // включить stash ← убрать конфликтующий механизм
+
+        lazyLoad: false,
+        deferLoadAfterSourceOpen: false,
+      }
+    );
    player.attachMediaElement(document.getElementById('v'))
    player.load()
    player.play()
--- a/app/src/handlers/demo-modbus.ts
+++ b/app/src/handlers/demo-modbus.ts
@@ -1,9 +1,9 @@
-import { postCloudIngest } from '../cloud-ingest.js'
 import { log } from '../helpers/log.js'
 import type { TopicHandler } from '../types.js'

 const EDGE = 'demo'
 const REGISTER_COUNT = 125
+const DEMO_CLOUD_INGEST_URL = process.env.DEMO_CLOUD_INGEST_URL as string

 type DemoModbusMetric = {
  edge: string
@@ -35,7 +35,14 @@ function toMetrics(values: number[], timestamp: string): DemoModbusMetric[] {

 async function postMetrics(metrics: DemoModbusMetric[]): Promise<void> {
  for (const metric of metrics) {
-    const response = await postCloudIngest(metric)
+    const response = await fetch(DEMO_CLOUD_INGEST_URL, {
+      method: 'POST',
+      headers: {
+        'content-type': 'application/json',
+        'x-api-key': process.env.CLOUD_INGEST_API_KEY as string,
+      },
+      body: JSON.stringify(metric),
+    })

    if (!response.ok) {
      const text = await response.text()
--- a/app/src/handlers/demo-plc.ts
+++ b/app/src/handlers/demo-plc.ts
@@ -1,9 +1,9 @@
-import { postCloudIngest } from '../cloud-ingest.js'
 import { log } from '../helpers/log.js'
 import { parseJson } from '../helpers/parse.js'
 import type { TopicHandler } from '../types.js'

 const EDGE = 'demo'
+const DEMO_CLOUD_INGEST_URL = process.env.DEMO_CLOUD_INGEST_URL as string

 type DemoPlcPayload = {
  tag: string
@@ -27,7 +27,14 @@ function toMetric(payload: DemoPlcPayload, timestamp: string): DemoPlcMetric {
 }

 async function postMetric(metric: DemoPlcMetric): Promise<void> {
-  const response = await postCloudIngest(metric)
+  const response = await fetch(DEMO_CLOUD_INGEST_URL, {
+    method: 'POST',
+    headers: {
+      'content-type': 'application/json',
+      'x-api-key': process.env.CLOUD_INGEST_API_KEY as string,
+    },
+    body: JSON.stringify(metric),
+  })

  if (!response.ok) {
    const text = await response.text()
--- a/app/src/handlers/edge5-video-v2.ts
+++ b/app/src/handlers/edge5-video-v2.ts
@@ -0,0 +1,21 @@
+import { log } from '../helpers/log.js'
+import { lastTopicSegment } from '../helpers/topic-match.js'
+import { onEdgeChunk } from '../stream/edge-chunk-relay.js'
+import type { TopicHandler } from '../types.js'
+
+const LOG_INTERVAL_MS = 10_000
+
+const lastLogByCamera = new Map<string, number>()
+
+export const handleEdge5VideoV2: TopicHandler = async (topic, payload) => {
+  const cameraId = lastTopicSegment(topic)
+
+  const now = Date.now()
+  const lastLog = lastLogByCamera.get(cameraId) ?? 0
+  if (now - lastLog >= LOG_INTERVAL_MS) {
+    log(`edge5.video.v2 [${cameraId}] ↑ ${payload.length}B`)
+    lastLogByCamera.set(cameraId, now)
+  }
+
+  onEdgeChunk(cameraId, payload);
+}
--- a/app/src/handlers/edge5-video.ts
+++ b/app/src/handlers/edge5-video.ts
@@ -2,7 +2,14 @@ import { log } from '../helpers/log.js'
 import { onEdgeChunk } from '../stream/edge-chunk-relay.js'
 import type { TopicHandler } from '../types.js'

+const LOG_INTERVAL_MS = 10_000
+let lastLog = 0
+
 export const handleEdge5Video: TopicHandler = async (_topic, payload) => {
-  log(`edge5.video ↑ ${payload.length}B`);
-  onEdgeChunk(payload);
-}
+  const now = Date.now()
+  if (now - lastLog >= LOG_INTERVAL_MS) {
+    log(`edge5.video ↑ ${payload.length}B`)
+    lastLog = now
+  }
+  onEdgeChunk('v1', payload);
+}
--- a/app/src/handlers/edge5cdab-modbus.ts
+++ b/app/src/handlers/edge5cdab-modbus.ts
@@ -10,7 +10,7 @@ type Edge5iiModbusMetric = {
  timestamp: string
  tag: string
  value: number
-}
+};

 function toFloatCDAB(reg0: number, reg1: number): number {
  const buf = Buffer.allocUnsafe(4)
@@ -35,11 +35,21 @@ function toMetrics(values: number[], timestamp: string): Edge5iiModbusMetric[] {
  const metrics: Edge5iiModbusMetric[] = []

  for (let i = 0; i < values.length; i += 2) {
+    const tag = `register-${String(i).padStart(3, '0')}`
+    const value = toFloatCDAB(values[i], values[i + 1])
+    if (isNaN(value)) {
+      log(
+        `${EDGE}.modbus: не удалось собрать число для ${tag}, ` +
+          `исходные values[${i}]=${values[i]}, values[${i + 1}]=${values[i + 1]}, пропуск`,
+      )
+      continue
+    };
+
    metrics.push({
      edge: EDGE,
      timestamp,
-      tag: `register-${String(i).padStart(3, '0')}`,
-      value: toFloatCDAB(values[i], values[i + 1]),
+      tag,
+      value,
    })
  }

@@ -60,9 +70,10 @@ async function postMetrics(metrics: Edge5iiModbusMetric[]): Promise<void> {
 }

 export const handle: TopicHandler = async (topic, payload) => {
+  const timestamp = new Date().toISOString()
  log(`${EDGE}.modbus.received ${topic} ${payload.length} bytes`)

  const values = parseValues(payload)
-  const metrics = toMetrics(values, new Date().toISOString())
+  const metrics = toMetrics(values, timestamp);
  await postMetrics(metrics)
 }
--- a/app/src/handlers/index.ts
+++ b/app/src/handlers/index.ts
@@ -4,6 +4,7 @@ import { handleDemoPlc } from './demo-plc.js';
 import { handleDemoModbus } from './demo-modbus.js';
 import { handleEdge5ModbusV2 } from './edge5-modbus-v2.js';
 import { handleEdge5Video } from './edge5-video.js';
+import { handleEdge5VideoV2 } from './edge5-video-v2.js';

 import { handle as interpretation } from './edge5i-modbus.js';
 import { handle as cdab} from './edge5cdab-modbus.js';
@@ -20,3 +21,4 @@ register(EDGE5_VIDEO_TOPIC, handleEdge5Video);

 register('data/edge5i/modbus', interpretation);
 register('data/edge5cdab/modbus', cdab);
+register('data/edge5/video/v2/+', handleEdge5VideoV2);
--- a/app/src/helpers/topic-match.ts
+++ b/app/src/helpers/topic-match.ts
@@ -0,0 +1,38 @@
+export function topicMatchesFilter(filter: string, topic: string): boolean {
+  const filterParts = filter.split('/')
+  const topicParts = topic.split('/')
+
+  let fi = 0
+  let ti = 0
+
+  while (fi < filterParts.length) {
+    const segment = filterParts[fi]
+
+    if (segment === '#') {
+      return fi === filterParts.length - 1
+    }
+
+    if (ti >= topicParts.length) {
+      return false
+    }
+
+    if (segment === '+') {
+      fi++
+      ti++
+      continue
+    }
+
+    if (segment !== topicParts[ti]) {
+      return false
+    }
+
+    fi++
+    ti++
+  }
+
+  return ti === topicParts.length
+}
+
+export function lastTopicSegment(topic: string): string {
+  return topic.split('/').at(-1) ?? ''
+}
--- 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'

--- a/app/src/registry.ts
+++ b/app/src/registry.ts
@@ -1,18 +1,37 @@
+import { topicMatchesFilter } from './helpers/topic-match.js'
 import type { TopicHandler } from './types.js'

-const handlers = new Map<string, TopicHandler>()
+const exactHandlers = new Map<string, TopicHandler>()
+const patternHandlers = new Map<string, TopicHandler>()
+
+function isPatternTopic(topic: string): boolean {
+  return topic.includes('+') || topic.includes('#')
+}

 export function register(topic: string, handler: TopicHandler): void {
+  const handlers = isPatternTopic(topic) ? patternHandlers : exactHandlers
+
  if (handlers.has(topic)) {
    throw new Error(`Handler already registered for topic: ${topic}`)
  }
  handlers.set(topic, handler)
 }

-export function getHandler(topic: string): TopicHandler | undefined {
-  return handlers.get(topic)
+export function resolveHandler(topic: string): TopicHandler | undefined {
+  const exact = exactHandlers.get(topic)
+  if (exact) {
+    return exact
+  }
+
+  for (const [filter, handler] of patternHandlers) {
+    if (topicMatchesFilter(filter, topic)) {
+      return handler
+    }
+  }
+
+  return undefined
 }

 export function getRegisteredTopics(): string[] {
-  return [...handlers.keys()]
+  return [...exactHandlers.keys(), ...patternHandlers.keys()]
 }
--- a/app/src/router.ts
+++ b/app/src/router.ts
@@ -1,10 +1,10 @@
-import { getHandler } from './registry.js'
+import { resolveHandler } from './registry.js'

 export async function routeMessage(
  topic: string,
  payload: Buffer,
 ): Promise<void> {
-  const handler = getHandler(topic)
+  const handler = resolveHandler(topic)

  if (!handler) {
    console.warn('mqtt.unhandled_topic', { topic })
--- a/app/src/stream/edge-chunk-relay.ts
+++ b/app/src/stream/edge-chunk-relay.ts
@@ -2,30 +2,40 @@ import type { IncomingMessage } from 'http'
 import { WebSocket, WebSocketServer } from 'ws'

 import { log } from '../helpers/log.js'
+import { lastTopicSegment } from '../helpers/topic-match.js'

 const wsPort = Number(process.env.WS_PORT)

 const wss = new WebSocketServer({ port: wsPort, perMessageDeflate: false })
-const clients = new Set<WebSocket>()
+
+const clientsByCamera = new Map<string, Set<WebSocket>>()
+
+function getClients(cameraId: string): Set<WebSocket> {
+  let clients = clientsByCamera.get(cameraId)
+  if (!clients) {
+    clients = new Set()
+    clientsByCamera.set(cameraId, clients)
+  }
+  return clients
+}

 wss.on('connection', (ws: WebSocket, req: IncomingMessage) => {
-  clients.add(ws)
-  log('ws.client_connected', {
-    clients: clients.size,
-    ip: req.socket.remoteAddress,
-  })
+  log(`ws.connection ${req.url}`);
+  const cameraId = req.url ? lastTopicSegment(req.url) : 'v1';
+
+  const clients = getClients(cameraId);
+  clients.add(ws);

  ws.on('close', () => {
    clients.delete(ws)
-    log('ws.client_closed', { clients: clients.size })
-  })
+    if (clients.size === 0) clientsByCamera.delete(cameraId)
+  });
+
 })

-wss.on('listening', () => {
-  log('ws.server_started', { port: wsPort })
-})
-
-export function onEdgeChunk(chunk: Buffer): void {
+export function onEdgeChunk(cameraId: string, chunk: Buffer): void {
+  const clients = clientsByCamera.get(cameraId)
+  if (!clients) return
  for (const client of clients) {
    if (client.readyState === WebSocket.OPEN) {
      client.send(chunk)
--- a/app/src/stream/ws-pipe-relay.ts
+++ 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<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 })
+})
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -3,25 +3,37 @@ services:
    build: ./mosquitto
    container_name: mosquitto
    ports:
-      - "1883:1883"
+      - "1883:1883" # external MQTT port for receiving data
    volumes:
      - mosquitto_data:/mosquitto/data
+    networks:
+      - default

  app:
    build: ./app
    container_name: mqtt-worker
    depends_on:
      - mosquitto
-    ports:
-      # - "80:80"
-      - "9090:9090"
+    expose:
+      - 80 
+      - 9090 
+      - 9091
    environment:
      MQTT_URL: mqtt://mosquitto:1883
-      HTTP_PORT: "80"
-      WS_PORT: "9090"
-      CLOUD_INGEST_URL: ${CLOUD_INGEST_URL:-https://demo.backend.drill.greact.ru/ingest}
-      CLOUD_INGEST_API_KEY: ${CLOUD_INGEST_API_KEY:-}
+      HTTP_PORT: 80 # http port for video previewiong page
+      WS_PORT: 9090 # WebSocket port for video streaming
+      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
+      - proxy

 volumes:
  mosquitto_data:
+
+networks:
+  proxy:
+    external: true
Author	SHA1	Message	Date
Nikita Biziaev	cd599a177b	ws-pipe-relay	2026-07-04 21:03:50 +05:00
Nikita Biziaev	a829a9df3d	video optimization/	2026-07-04 18:57:41 +05:00
Nikita Biziaev	eb001771b5	WebSocketServer.clientsByCamera	2026-07-04 18:09:16 +05:00
Nikita Biziaev	e0437edf18	data/edge5/video/v2/+	2026-07-04 17:26:30 +05:00
Nikita Biziaev	f61fd87bac	fit video	2026-07-03 17:38:23 +05:00
Nikita Biziaev	80772c8751	custom domain name	2026-07-03 17:30:05 +05:00
Nikita Biziaev	3eb9680ef8	networks comments	2026-07-03 17:16:31 +05:00
Nikita Biziaev	495651379a	video preview page	2026-07-03 17:07:55 +05:00
Nikita Biziaev	11a4f9d03a	app/public/index.html	2026-07-03 16:51:55 +05:00
Nikita Biziaev	9988281368	exceeded log	2026-07-03 13:34:52 +05:00
Nikita Biziaev	60bde31a7b	cdab validate numbers	2026-07-03 12:13:54 +05:00
Nikita Biziaev	5b53f4ed12	Merge branch 'dev'	2026-07-03 12:06:03 +05:00
Nikita Biziaev	337497171a	stringify	2026-07-03 12:03:39 +05:00
Никита Бизяев	d65bf84a60	Merge pull request 'добавлена поддержка dev окружения' (#10 ) from dev into main Reviewed-on: #10	2026-07-03 06:38:41 +00:00
Первов Артем	64dde77f7b	добавлена поддержка dev окружения	2026-07-02 22:39:59 +03:00