Обновил отправку тегов с массива на построчный формат #8

Merged
NBizyaev merged 2 commits from edge5modbus into main 2026-06-22 06:45:44 +00:00
4 changed files with 39 additions and 19 deletions

View File

@@ -13,12 +13,12 @@ type DemoPlcPayload = {
type DemoPlcMetric = { type DemoPlcMetric = {
edge: string edge: string
timestamp: number timestamp: string
tag: string tag: string
value: number value: number
} }
function toMetric(payload: DemoPlcPayload, timestamp: number): DemoPlcMetric { function toMetric(payload: DemoPlcPayload, timestamp: string): DemoPlcMetric {
return { return {
edge: EDGE, edge: EDGE,
timestamp, timestamp,
@@ -28,7 +28,7 @@ function toMetric(payload: DemoPlcPayload, timestamp: number): DemoPlcMetric {
} }
async function postMetric(metric: DemoPlcMetric): Promise<void> { async function postMetric(metric: DemoPlcMetric): Promise<void> {
const response = await postCloudIngest([metric]) const response = await postCloudIngest(metric)
if (!response.ok) { if (!response.ok) {
const text = await response.text() const text = await response.text()
@@ -42,6 +42,6 @@ export const handleDemoPlc: TopicHandler = async (topic, payload) => {
console.log('demo.plc.received', { topic, bytes: payload.length }) console.log('demo.plc.received', { topic, bytes: payload.length })
const value = parseJson<DemoPlcPayload>(payload) const value = parseJson<DemoPlcPayload>(payload)
const metric = toMetric(value, Date.now()) const metric = toMetric(value, new Date().toISOString())
await postMetric(metric) await postMetric(metric)
} }

View File

@@ -22,20 +22,38 @@ function assertRegisterCount(values: number[]): void {
} }
async function postMetrics(metrics: MappedMetric[]): Promise<void> { async function postMetrics(metrics: MappedMetric[]): Promise<void> {
const response = await postCloudIngest(metrics) const errors: Error[] = []
for (const metric of metrics) {
try {
const response = await postCloudIngest(metric)
if (!response.ok) { if (!response.ok) {
const text = await response.text() const text = await response.text()
throw new Error(text) throw new Error(text || `${response.status} ${response.statusText}`)
}
} catch (error) {
const message = error instanceof Error ? error.message : String(error)
errors.push(new Error(`${metric.tag}: ${message}`))
console.error('edge5.modbus.v2.metric.failed', metric.tag, message)
}
} }
console.log('edge5.modbus.v2.posted', { count: metrics.length }) console.log('edge5.modbus.v2.posted', {
total: metrics.length,
succeeded: metrics.length - errors.length,
failed: errors.length,
})
if (errors.length) {
throw new Error(`Failed to post ${errors.length}/${metrics.length} edge5 modbus v2 metrics`)
}
} }
export const handleEdge5ModbusV2: TopicHandler = async (topic, payload) => { export const handleEdge5ModbusV2: TopicHandler = async (topic, payload) => {
console.log('edge5.modbus.v2.received', { topic, bytes: payload.length }) console.log('edge5.modbus.v2.received', { topic, bytes: payload.length })
const values = parseValues(payload) const values = parseValues(payload)
const metrics = mapper.mapValues(values, Date.now()) const metrics = mapper.mapValues(values, new Date().toISOString())
await postMetrics(metrics) await postMetrics(metrics)
} }

View File

@@ -8,7 +8,7 @@ const REGISTER_COUNT = 125
type Edge5ModbusMetric = { type Edge5ModbusMetric = {
edge: string edge: string
timestamp: number timestamp: string
tag: string tag: string
value: number value: number
} }
@@ -25,7 +25,7 @@ function assertRegisterCount(values: number[]): void {
} }
} }
function toMetrics(values: number[], timestamp: number): Edge5ModbusMetric[] { function toMetrics(values: number[], timestamp: string): Edge5ModbusMetric[] {
return values.map((value, index) => ({ return values.map((value, index) => ({
edge: EDGE, edge: EDGE,
timestamp, timestamp,
@@ -35,12 +35,14 @@ function toMetrics(values: number[], timestamp: number): Edge5ModbusMetric[] {
} }
async function postMetrics(metrics: Edge5ModbusMetric[]): Promise<void> { async function postMetrics(metrics: Edge5ModbusMetric[]): Promise<void> {
const response = await postCloudIngest(metrics) for (const metric of metrics) {
const response = await postCloudIngest(metric)
if (!response.ok) { if (!response.ok) {
const text = await response.text() const text = await response.text()
throw new Error(text) throw new Error(text)
} }
}
console.log('edge5.modbus.posted', { count: metrics.length }) console.log('edge5.modbus.posted', { count: metrics.length })
} }
@@ -49,6 +51,6 @@ export const handleEdge5Modbus: TopicHandler = async (topic, payload) => {
console.log('edge5.modbus.received', { topic, bytes: payload.length }) console.log('edge5.modbus.received', { topic, bytes: payload.length })
const values = parseValues(payload) const values = parseValues(payload)
const metrics = toMetrics(values, Date.now()) const metrics = toMetrics(values, new Date().toISOString())
await postMetrics(metrics) await postMetrics(metrics)
} }

View File

@@ -2,7 +2,7 @@ import { readFileSync } from 'node:fs'
export type MappedMetric = { export type MappedMetric = {
edge: string edge: string
timestamp: number timestamp: string
tag: string tag: string
value: number value: number
} }
@@ -14,7 +14,7 @@ type MappingConfig = {
export type MetricMapper = { export type MetricMapper = {
edge: string edge: string
mapValues: (values: number[], timestamp: number) => MappedMetric[] mapValues: (values: number[], timestamp: string) => MappedMetric[]
} }
type MetricMapperOptions = { type MetricMapperOptions = {