data/edge5/video/v2/+

app/src/handlers/edge5-video-v2.ts

@@ -0,0 +1,23 @@
+import { log } from '../helpers/log.js'
+import { lastTopicSegment } from '../helpers/topic-match.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)
+
+  if (!cameraId) {
+    console.warn('edge5.video.v2.invalid_topic', { topic })
+    return
+  }
+
+  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)
+  }
+}

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

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

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()]
 }

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