NBizyaev/mqtt-ingest
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Compare commits

base: NBizyaev:d7bfaa678e0bd2e594ca4d0ded4ccd91c839e4e5
Branches Tags
NBizyaev:main NBizyaev:dev-artem NBizyaev:wss-ingest NBizyaev:dev NBizyaev:edge5modbus
...
compare: NBizyaev:wss-ingest
Branches Tags
NBizyaev:dev-artem NBizyaev:main NBizyaev:wss-ingest NBizyaev:dev NBizyaev:edge5modbus

15 Commits
d7bfaa678e ... wss-ingest

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
15 changed files with 300 additions and 43 deletions
Expand all files Collapse all files

4
app/Dockerfile
View File

@@ -10,5 +10,9 @@ COPY src ./src
RUN npm run build && npm prune --omit=dev 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 ENV NODE_ENV=production
CMD ["node", "dist/index.js"] CMD ["node", "dist/index.js"]

34
app/public/index.html
View File

@@ -4,15 +4,39 @@
<meta charset="utf-8" /> <meta charset="utf-8" />
<title>edge5 video relay</title> <title>edge5 video relay</title>
<script src="https://cdn.jsdelivr.net/npm/mpegts.js/dist/mpegts.js"></script> <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> </head>
<body> <body>
<video id="v" controls autoplay muted style="width:100%"></video> <video id="v" controls autoplay muted></video>
<script> <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( const player = mpegts.createPlayer(
{ type: 'mpegts', isLive: true, url: wsUrl }, { type: 'mpegts', isLive: true, url: wsUrl, hasAudio: false },
{ liveBufferLatencyChasing: true, liveBufferLatencyMaxLatency: 1.5 } {
) 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.attachMediaElement(document.getElementById('v'))
player.load() player.load()
player.play() player.play()

11
app/src/handlers/demo-modbus.ts
View File

@@ -1,9 +1,9 @@
import { postCloudIngest } from '../cloud-ingest.js'
import { log } from '../helpers/log.js' import { log } from '../helpers/log.js'
import type { TopicHandler } from '../types.js' import type { TopicHandler } from '../types.js'
const EDGE = 'demo' const EDGE = 'demo'
const REGISTER_COUNT = 125 const REGISTER_COUNT = 125
const DEMO_CLOUD_INGEST_URL = process.env.DEMO_CLOUD_INGEST_URL as string
type DemoModbusMetric = { type DemoModbusMetric = {
edge: string edge: string
@@ -35,7 +35,14 @@ function toMetrics(values: number[], timestamp: string): DemoModbusMetric[] {
async function postMetrics(metrics: DemoModbusMetric[]): Promise<void> { async function postMetrics(metrics: DemoModbusMetric[]): Promise<void> {
for (const metric of metrics) { 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) { if (!response.ok) {
const text = await response.text() const text = await response.text()

11
app/src/handlers/demo-plc.ts
View File

@@ -1,9 +1,9 @@
import { postCloudIngest } from '../cloud-ingest.js'
import { log } from '../helpers/log.js' import { log } from '../helpers/log.js'
import { parseJson } from '../helpers/parse.js' import { parseJson } from '../helpers/parse.js'
import type { TopicHandler } from '../types.js' import type { TopicHandler } from '../types.js'
const EDGE = 'demo' const EDGE = 'demo'
const DEMO_CLOUD_INGEST_URL = process.env.DEMO_CLOUD_INGEST_URL as string
type DemoPlcPayload = { type DemoPlcPayload = {
tag: string tag: string
@@ -27,7 +27,14 @@ function toMetric(payload: DemoPlcPayload, timestamp: string): DemoPlcMetric {
} }
async function postMetric(metric: DemoPlcMetric): Promise<void> { 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) { if (!response.ok) {
const text = await response.text() const text = await response.text()

21
app/src/handlers/edge5-video-v2.ts Normal file
View File

@@ -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);
}

11
app/src/handlers/edge5-video.ts
View File

@@ -2,7 +2,14 @@ import { log } from '../helpers/log.js'
import { onEdgeChunk } from '../stream/edge-chunk-relay.js' import { onEdgeChunk } from '../stream/edge-chunk-relay.js'
import type { TopicHandler } from '../types.js' import type { TopicHandler } from '../types.js'
const LOG_INTERVAL_MS = 10_000
let lastLog = 0
export const handleEdge5Video: TopicHandler = async (_topic, payload) => { export const handleEdge5Video: TopicHandler = async (_topic, payload) => {
log(`edge5.video ↑ ${payload.length}B`); const now = Date.now()
onEdgeChunk(payload); if (now - lastLog >= LOG_INTERVAL_MS) {
log(`edge5.video ↑ ${payload.length}B`)
lastLog = now
}
onEdgeChunk('v1', payload);
} }

19
app/src/handlers/edge5cdab-modbus.ts
View File

@@ -10,7 +10,7 @@ type Edge5iiModbusMetric = {
timestamp: string timestamp: string
tag: string tag: string
value: number value: number
} };
function toFloatCDAB(reg0: number, reg1: number): number { function toFloatCDAB(reg0: number, reg1: number): number {
const buf = Buffer.allocUnsafe(4) const buf = Buffer.allocUnsafe(4)
@@ -35,11 +35,21 @@ function toMetrics(values: number[], timestamp: string): Edge5iiModbusMetric[] {
const metrics: Edge5iiModbusMetric[] = [] const metrics: Edge5iiModbusMetric[] = []
for (let i = 0; i < values.length; i += 2) { 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({ metrics.push({
edge: EDGE, edge: EDGE,
timestamp, timestamp,
tag: `register-${String(i).padStart(3, '0')}`, tag,
value: toFloatCDAB(values[i], values[i + 1]), value,
}) })
} }
@@ -60,9 +70,10 @@ async function postMetrics(metrics: Edge5iiModbusMetric[]): Promise<void> {
} }
export const handle: TopicHandler = async (topic, payload) => { export const handle: TopicHandler = async (topic, payload) => {
const timestamp = new Date().toISOString()
log(`${EDGE}.modbus.received ${topic} ${payload.length} bytes`) log(`${EDGE}.modbus.received ${topic} ${payload.length} bytes`)
const values = parseValues(payload) const values = parseValues(payload)
const metrics = toMetrics(values, new Date().toISOString()) const metrics = toMetrics(values, timestamp);
await postMetrics(metrics) await postMetrics(metrics)
} }

2
app/src/handlers/index.ts
View File

@@ -4,6 +4,7 @@ import { handleDemoPlc } from './demo-plc.js';
import { handleDemoModbus } from './demo-modbus.js'; import { handleDemoModbus } from './demo-modbus.js';
import { handleEdge5ModbusV2 } from './edge5-modbus-v2.js'; import { handleEdge5ModbusV2 } from './edge5-modbus-v2.js';
import { handleEdge5Video } from './edge5-video.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 interpretation } from './edge5i-modbus.js';
import { handle as cdab} from './edge5cdab-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/edge5i/modbus', interpretation);
register('data/edge5cdab/modbus', cdab); register('data/edge5cdab/modbus', cdab);
register('data/edge5/video/v2/+', handleEdge5VideoV2);

38
app/src/helpers/topic-match.ts Normal file
View File

@@ -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) ?? ''
}

1
app/src/index.ts
View File

@@ -2,6 +2,7 @@ import './handlers/index.js'
import './http/server.js' import './http/server.js'
import './mqtt.js' import './mqtt.js'
import './stream/edge-chunk-relay.js' import './stream/edge-chunk-relay.js'
import './stream/ws-pipe-relay.js'
import { log } from './helpers/log.js' import { log } from './helpers/log.js'

27
app/src/registry.ts
View File

@@ -1,18 +1,37 @@
import { topicMatchesFilter } from './helpers/topic-match.js'
import type { TopicHandler } from './types.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 { export function register(topic: string, handler: TopicHandler): void {
const handlers = isPatternTopic(topic) ? patternHandlers : exactHandlers
if (handlers.has(topic)) { if (handlers.has(topic)) {
throw new Error(`Handler already registered for topic: ${topic}`) throw new Error(`Handler already registered for topic: ${topic}`)
} }
handlers.set(topic, handler) handlers.set(topic, handler)
} }
export function getHandler(topic: string): TopicHandler | undefined { export function resolveHandler(topic: string): TopicHandler | undefined {
return handlers.get(topic) 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[] { export function getRegisteredTopics(): string[] {
return [...handlers.keys()] return [...exactHandlers.keys(), ...patternHandlers.keys()]
} }

4
app/src/router.ts
View File

@@ -1,10 +1,10 @@
import { getHandler } from './registry.js' import { resolveHandler } from './registry.js'
export async function routeMessage( export async function routeMessage(
topic: string, topic: string,
payload: Buffer, payload: Buffer,
): Promise<void> { ): Promise<void> {
const handler = getHandler(topic) const handler = resolveHandler(topic)
if (!handler) { if (!handler) {
console.warn('mqtt.unhandled_topic', { topic }) console.warn('mqtt.unhandled_topic', { topic })

36
app/src/stream/edge-chunk-relay.ts
View File

@@ -2,30 +2,40 @@ import type { IncomingMessage } from 'http'
import { WebSocket, WebSocketServer } from 'ws' import { WebSocket, WebSocketServer } from 'ws'
import { log } from '../helpers/log.js' import { log } from '../helpers/log.js'
import { lastTopicSegment } from '../helpers/topic-match.js'
const wsPort = Number(process.env.WS_PORT) const wsPort = Number(process.env.WS_PORT)
const wss = new WebSocketServer({ port: wsPort, perMessageDeflate: false }) 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) => { wss.on('connection', (ws: WebSocket, req: IncomingMessage) => {
clients.add(ws) log(`ws.connection ${req.url}`);
log('ws.client_connected', { const cameraId = req.url ? lastTopicSegment(req.url) : 'v1';
clients: clients.size,
ip: req.socket.remoteAddress, const clients = getClients(cameraId);
}) clients.add(ws);
ws.on('close', () => { ws.on('close', () => {
clients.delete(ws) clients.delete(ws)
log('ws.client_closed', { clients: clients.size }) if (clients.size === 0) clientsByCamera.delete(cameraId)
}) });
}) })
wss.on('listening', () => { export function onEdgeChunk(cameraId: string, chunk: Buffer): void {
log('ws.server_started', { port: wsPort }) const clients = clientsByCamera.get(cameraId)
}) if (!clients) return
export function onEdgeChunk(chunk: Buffer): void {
for (const client of clients) { for (const client of clients) {
if (client.readyState === WebSocket.OPEN) { if (client.readyState === WebSocket.OPEN) {
client.send(chunk) client.send(chunk)

94
app/src/stream/ws-pipe-relay.ts Normal file
View File

@@ -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 })
})

28
docker-compose.yml
View File

@@ -3,25 +3,37 @@ services:
build: ./mosquitto build: ./mosquitto
container_name: mosquitto container_name: mosquitto
ports: ports:
- "1883:1883" - "1883:1883" # external MQTT port for receiving data
volumes: volumes:
- mosquitto_data:/mosquitto/data - mosquitto_data:/mosquitto/data
networks:
- default
app: app:
build: ./app build: ./app
container_name: mqtt-worker container_name: mqtt-worker
depends_on: depends_on:
- mosquitto - mosquitto
ports: expose:
# - "80:80" - 80
- "9090:9090" - 9090
- 9091
environment: environment:
MQTT_URL: mqtt://mosquitto:1883 MQTT_URL: mqtt://mosquitto:1883
HTTP_PORT: "80" HTTP_PORT: 80 # http port for video previewiong page
WS_PORT: "9090" 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}
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} EDGE5_MODBUS_V2_MAPPING_FILE: ${EDGE5_MODBUS_V2_MAPPING_FILE:-/app/src/mappings/edge5-modbus.json}
networks:
- default
- proxy
volumes: volumes:
mosquitto_data: mosquitto_data:
networks:
proxy:
external: true