Compare commits
23 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| fe9e7b5438 | |||
| 3c94cd92c1 | |||
| 5b97022116 | |||
| 7c67b9e34a | |||
| 3841faa810 | |||
| 5d18bb7965 | |||
| 37539469de | |||
| 45a31c7210 | |||
| 03d2710fdb | |||
| 52beffe377 | |||
| cd599a177b | |||
| a829a9df3d | |||
| eb001771b5 | |||
| e0437edf18 | |||
| f61fd87bac | |||
| 80772c8751 | |||
| 3eb9680ef8 | |||
| 495651379a | |||
| 11a4f9d03a | |||
| 9988281368 | |||
| 60bde31a7b | |||
| 5b53f4ed12 | |||
| d65bf84a60 |
@@ -10,5 +10,7 @@ COPY src ./src
|
||||
|
||||
RUN npm run build && npm prune --omit=dev
|
||||
|
||||
COPY assets ./assets
|
||||
|
||||
ENV NODE_ENV=production
|
||||
CMD ["node", "dist/index.js"]
|
||||
|
||||
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).
|
||||
45
app/assets/index.html
Normal file
45
app/assets/index.html
Normal file
@@ -0,0 +1,45 @@
|
||||
<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
<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></video>
|
||||
<script>
|
||||
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, 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()
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -1,21 +0,0 @@
|
||||
<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
<meta charset="utf-8" />
|
||||
<title>edge5 video relay</title>
|
||||
<script src="https://cdn.jsdelivr.net/npm/mpegts.js/dist/mpegts.js"></script>
|
||||
</head>
|
||||
<body>
|
||||
<video id="v" controls autoplay muted style="width:100%"></video>
|
||||
<script>
|
||||
const wsUrl = `ws://${location.hostname}:9090`
|
||||
const player = mpegts.createPlayer(
|
||||
{ type: 'mpegts', isLive: true, url: wsUrl },
|
||||
{ liveBufferLatencyChasing: true, liveBufferLatencyMaxLatency: 1.5 }
|
||||
)
|
||||
player.attachMediaElement(document.getElementById('v'))
|
||||
player.load()
|
||||
player.play()
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -1,4 +1,5 @@
|
||||
export const cloudIngestUrl = process.env.CLOUD_INGEST_URL as string
|
||||
export const demoCloudIngestUrl = process.env.DEMO_CLOUD_INGEST_URL as string
|
||||
|
||||
export function postCloudIngest(body: unknown): Promise<Response> {
|
||||
return fetch(cloudIngestUrl, {
|
||||
@@ -10,3 +11,14 @@ export function postCloudIngest(body: unknown): Promise<Response> {
|
||||
body: JSON.stringify(body),
|
||||
})
|
||||
}
|
||||
|
||||
export function postDemoCloudIngest(body: unknown): Promise<Response> {
|
||||
return fetch(demoCloudIngestUrl, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'content-type': 'application/json',
|
||||
'x-api-key': process.env.CLOUD_INGEST_API_KEY as string,
|
||||
},
|
||||
body: JSON.stringify(body),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import { dirname, join } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
|
||||
import { postCloudIngest } from '../cloud-ingest.js'
|
||||
import { log } from '../helpers/log.js'
|
||||
import { createMetricMapper, type MappedMetric } from '../mapping.js'
|
||||
@@ -5,8 +8,13 @@ import type { TopicHandler } from '../types.js'
|
||||
|
||||
const REGISTER_COUNT = 125
|
||||
|
||||
const defaultMappingFile = join(
|
||||
dirname(fileURLToPath(import.meta.url)),
|
||||
'../../assets/edge5-modbus.json',
|
||||
)
|
||||
|
||||
const mapper = createMetricMapper({
|
||||
filePath: process.env.EDGE5_MODBUS_V2_MAPPING_FILE as string,
|
||||
filePath: process.env.EDGE5_MODBUS_V2_MAPPING_FILE ?? defaultMappingFile,
|
||||
})
|
||||
|
||||
function parseValues(payload: Buffer): number[] {
|
||||
|
||||
113
app/src/handlers/edge5-modbus-v3.ts
Normal file
113
app/src/handlers/edge5-modbus-v3.ts
Normal file
@@ -0,0 +1,113 @@
|
||||
import { postCloudIngest, postDemoCloudIngest } from '../cloud-ingest.js'
|
||||
import { log } from '../helpers/log.js'
|
||||
import type { TopicHandler } from '../types.js'
|
||||
|
||||
const EDGE = 'edge5-v3'
|
||||
const REGISTER_COUNT = 24
|
||||
|
||||
// Фиксируем интерпретацию
|
||||
const TAGS = [
|
||||
'edge5-v3-wk',
|
||||
'edge5-v3-pin',
|
||||
'edge5-v3-hk',
|
||||
'edge5-v3-mk',
|
||||
'edge5-v3-mrot',
|
||||
'edge5-v3-h',
|
||||
'edge5-v3-vw',
|
||||
'edge5-v3-pdk',
|
||||
'edge5-v3-h2s',
|
||||
'edge5-v3-t',
|
||||
'edge5-v3-nrot',
|
||||
'edge5-v3-vsp',
|
||||
] as const
|
||||
|
||||
// из них выводим на дашборд семь виджетов
|
||||
const CLOUD_INGEST_TAGS = new Set<string>([
|
||||
'edge5-v3-wk',
|
||||
'edge5-v3-hk',
|
||||
'edge5-v3-vw',
|
||||
'edge5-v3-pdk',
|
||||
'edge5-v3-h2s',
|
||||
'edge5-v3-nrot',
|
||||
'edge5-v3-vsp',
|
||||
])
|
||||
|
||||
type Edge5ModbusV3Metric = {
|
||||
edge: string
|
||||
timestamp: string
|
||||
tag: string
|
||||
value: number | null
|
||||
}
|
||||
|
||||
function toFloatCDAB(reg0: number, reg1: number): number | null {
|
||||
const buf = Buffer.allocUnsafe(4)
|
||||
buf.writeUInt16BE(reg1, 0)
|
||||
buf.writeUInt16BE(reg0, 2)
|
||||
return buf.readFloatBE(0)
|
||||
}
|
||||
|
||||
function parseValues(payload: Buffer): number[] {
|
||||
const values = JSON.parse(payload.toString('utf8')) as number[]
|
||||
assertRegisterCount(values)
|
||||
return values
|
||||
}
|
||||
|
||||
function assertRegisterCount(values: number[]): void {
|
||||
if (values.length !== REGISTER_COUNT) {
|
||||
throw new Error(`Expected ${REGISTER_COUNT} modbus values, got ${values.length}`)
|
||||
}
|
||||
}
|
||||
|
||||
function toMetrics(values: number[], timestamp: string): Edge5ModbusV3Metric[] {
|
||||
const metrics: Edge5ModbusV3Metric[] = []
|
||||
|
||||
for (let i = 0; i < values.length; i += 2) {
|
||||
const tag = TAGS[i / 2]
|
||||
const value = toFloatCDAB(values[i], values[i + 1]); // null is possible to send
|
||||
|
||||
metrics.push({
|
||||
edge: EDGE,
|
||||
timestamp,
|
||||
tag,
|
||||
value,
|
||||
})
|
||||
}
|
||||
|
||||
return metrics
|
||||
}
|
||||
|
||||
async function postMetrics(metrics: Edge5ModbusV3Metric[]): Promise<void> {
|
||||
const toPost = metrics.filter((metric) => CLOUD_INGEST_TAGS.has(metric.tag))
|
||||
const errors: Error[] = []
|
||||
|
||||
for (const metric of toPost) {
|
||||
try {
|
||||
const response = await postCloudIngest(metric)
|
||||
|
||||
if (!response.ok) {
|
||||
const text = await response.text()
|
||||
throw new Error(text || `${response.status} ${response.statusText}`)
|
||||
}
|
||||
|
||||
void postDemoCloudIngest(metric); // для демо-системы
|
||||
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error)
|
||||
errors.push(new Error(`${metric.tag}: ${message}`))
|
||||
console.error('edge5.modbus.v3.metric.failed', metric.tag, message)
|
||||
}
|
||||
}
|
||||
|
||||
log(`edge5.modbus.v3.data ↑ ${toPost.length} tags`);
|
||||
|
||||
if (errors.length) {
|
||||
throw new Error(`Failed to post ${errors.length}/${toPost.length} edge5 modbus v3 metrics`)
|
||||
}
|
||||
}
|
||||
|
||||
export const handleEdge5ModbusV3: TopicHandler = async (topic, payload) => {
|
||||
const timestamp = new Date().toISOString();
|
||||
const values = parseValues(payload);
|
||||
const metrics = toMetrics(values, timestamp);
|
||||
await postMetrics(metrics);
|
||||
}
|
||||
21
app/src/handlers/edge5-video-v2.ts
Normal file
21
app/src/handlers/edge5-video-v2.ts
Normal 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);
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
@@ -9,10 +9,10 @@ type Edge5iiModbusMetric = {
|
||||
edge: string
|
||||
timestamp: string
|
||||
tag: string
|
||||
value: number
|
||||
value: number | null
|
||||
};
|
||||
|
||||
function toFloatCDAB(reg0: number, reg1: number): number {
|
||||
function toFloatCDAB(reg0: number, reg1: number): number | null {
|
||||
const buf = Buffer.allocUnsafe(4)
|
||||
buf.writeUInt16BE(reg1, 0)
|
||||
buf.writeUInt16BE(reg0, 2)
|
||||
@@ -35,11 +35,14 @@ 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]); // null is possible to send
|
||||
|
||||
metrics.push({
|
||||
edge: EDGE,
|
||||
timestamp,
|
||||
tag: `register-${String(i).padStart(3, '0')}`,
|
||||
value: toFloatCDAB(values[i], values[i + 1]),
|
||||
tag,
|
||||
value,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -60,10 +63,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())
|
||||
log(`${EDGE}.modbus.prepared ${JSON.stringify(metrics)}`)
|
||||
const metrics = toMetrics(values, timestamp);
|
||||
await postMetrics(metrics)
|
||||
}
|
||||
|
||||
@@ -3,7 +3,9 @@ import { register } from '../registry.js';
|
||||
import { handleDemoPlc } from './demo-plc.js';
|
||||
import { handleDemoModbus } from './demo-modbus.js';
|
||||
import { handleEdge5ModbusV2 } from './edge5-modbus-v2.js';
|
||||
import { handleEdge5ModbusV3 } from './edge5-modbus-v3.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 +22,5 @@ register(EDGE5_VIDEO_TOPIC, handleEdge5Video);
|
||||
|
||||
register('data/edge5i/modbus', interpretation);
|
||||
register('data/edge5cdab/modbus', cdab);
|
||||
register('data/edge5/video/v2/+', handleEdge5VideoV2);
|
||||
register('data/edge5/modbus/v3', handleEdge5ModbusV3);
|
||||
|
||||
38
app/src/helpers/topic-match.ts
Normal file
38
app/src/helpers/topic-match.ts
Normal 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) ?? ''
|
||||
}
|
||||
@@ -6,8 +6,8 @@ import { fileURLToPath } from 'node:url'
|
||||
import { log } from '../helpers/log.js'
|
||||
|
||||
const httpPort = Number(process.env.HTTP_PORT)
|
||||
const publicDir = join(dirname(fileURLToPath(import.meta.url)), '../../public')
|
||||
const indexPath = join(publicDir, 'index.html')
|
||||
const assetsDir = join(dirname(fileURLToPath(import.meta.url)), '../../assets')
|
||||
const indexPath = join(assetsDir, 'index.html')
|
||||
|
||||
createServer(async (req, res) => {
|
||||
const path = req.url?.split('?')[0]
|
||||
|
||||
@@ -19,7 +19,7 @@ export type MetricMapper = {
|
||||
|
||||
type MetricMapperOptions = {
|
||||
// filePath contract:
|
||||
// - built-in mappings are passed as URL via new URL('../mappings/*.json', import.meta.url);
|
||||
// - built-in mappings live in app/assets/*.json (resolved relative to dist/ at runtime);
|
||||
// - env overrides are expected as absolute paths inside the runtime container.
|
||||
filePath: string | URL
|
||||
}
|
||||
|
||||
@@ -6,7 +6,12 @@ import { routeMessage } from './router.js'
|
||||
|
||||
const mqttUrl = process.env.MQTT_URL as string
|
||||
|
||||
export const mqttClient = mqtt.connect(mqttUrl)
|
||||
export const mqttClient = mqtt.connect(mqttUrl, {
|
||||
clean: true,
|
||||
queueQoSZero: false,
|
||||
reconnectPeriod: 1000,
|
||||
connectTimeout: 10000,
|
||||
});
|
||||
|
||||
mqttClient.on('connect', () => {
|
||||
log('mqtt.connected', { url: mqttUrl })
|
||||
|
||||
@@ -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()]
|
||||
}
|
||||
|
||||
@@ -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 })
|
||||
|
||||
@@ -2,30 +2,41 @@ 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}`);
|
||||
req.socket.setNoDelay(true)
|
||||
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)
|
||||
|
||||
@@ -3,26 +3,35 @@ 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
|
||||
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:-}
|
||||
DEMO_CLOUD_INGEST_URL: ${DEMO_CLOUD_INGEST_URL:-}
|
||||
EDGE5_MODBUS_V2_MAPPING_FILE: ${EDGE5_MODBUS_V2_MAPPING_FILE:-/app/src/mappings/edge5-modbus.json}
|
||||
HTTP_PORT: 80 # http port for video previewing page
|
||||
WS_PORT: 9090 # WebSocket port for video streaming
|
||||
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/assets/edge5-modbus.json}
|
||||
networks:
|
||||
- default
|
||||
- proxy
|
||||
|
||||
volumes:
|
||||
mosquitto_data:
|
||||
|
||||
networks:
|
||||
proxy:
|
||||
external: true
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
listener 1883
|
||||
allow_anonymous true
|
||||
|
||||
set_tcp_nodelay true # выключает Nagle на всех сокетах брокера
|
||||
|
||||
persistence true
|
||||
persistence_location /mosquitto/data/
|
||||
|
||||
|
||||
Reference in New Issue
Block a user