#!/usr/bin/env node const fs = require('fs'); const path = require('path'); const { TcpService } = require('./tcp-service'); const { MqttService } = require('./mqtt-service'); const { AliyunService } = require('./aliyun-service'); const { MetricAggregator } = require('./metric-aggregator'); const { getDefaultConfigPath } = require('./runtime-paths'); const LOG_LEVELS = { debug: 10, info: 20, warn: 30, error: 40, }; // 统一生成本地日志时间,兼顾控制台和文件日志格式。 function formatLocalTimestamp(date = new Date()) { const year = date.getFullYear(); const month = String(date.getMonth() + 1).padStart(2, '0'); const day = String(date.getDate()).padStart(2, '0'); const hour = String(date.getHours()).padStart(2, '0'); const minute = String(date.getMinutes()).padStart(2, '0'); const second = String(date.getSeconds()).padStart(2, '0'); const millisecond = String(date.getMilliseconds()).padStart(3, '0'); return `${year}-${month}-${day} ${hour}:${minute}:${second}.${millisecond}`; } function normalizeLogLevel(level) { const normalized = String(level || 'info').trim().toLowerCase(); return LOG_LEVELS[normalized] ? normalized : 'info'; } function createLogger(loggingConfig = {}) { const settings = { enabled: loggingConfig.enabled !== false, console: loggingConfig.console !== false, dir: loggingConfig.dir || path.join(process.cwd(), 'logs'), filePrefix: loggingConfig.filePrefix || 'jhm-service', level: normalizeLogLevel(loggingConfig.level), }; let currentDate = ''; let currentFilePath = ''; function getDatePart(date = new Date()) { const year = date.getFullYear(); const month = String(date.getMonth() + 1).padStart(2, '0'); const day = String(date.getDate()).padStart(2, '0'); return `${year}${month}${day}`; } function ensureLogFilePath() { if (!settings.enabled) { return null; } const today = getDatePart(); if (currentFilePath && currentDate === today) { return currentFilePath; } fs.mkdirSync(settings.dir, { recursive: true }); currentDate = today; currentFilePath = path.join(settings.dir, `${settings.filePrefix}-${today}.log`); return currentFilePath; } function shouldWrite(level) { return LOG_LEVELS[level] >= LOG_LEVELS[settings.level]; } function format(level, message) { return `[${formatLocalTimestamp()}] [${level.toUpperCase()}] ${message}`; } function write(level, message) { const normalizedLevel = normalizeLogLevel(level); if (!shouldWrite(normalizedLevel)) { return; } const line = format(normalizedLevel, message); if (settings.console) { if (normalizedLevel === 'error') { console.error(line); } else if (normalizedLevel === 'warn') { console.warn(line); } else { console.log(line); } } const logFilePath = ensureLogFilePath(); if (logFilePath) { try { fs.appendFileSync(logFilePath, `${line}\n`, 'utf8'); } catch (error) { console.error(`[LOGGER] 写入日志文件失败: ${error.message}`); } } } return { debug(message) { write('debug', message); }, info(message) { write('info', message); }, warn(message) { write('warn', message); }, error(message) { write('error', message); }, getLogFilePath() { ensureLogFilePath(); return currentFilePath; }, async close() { currentDate = ''; }, }; } function parseArgs(argv) { const options = { configPath: getDefaultConfigPath(), }; for (let index = 2; index < argv.length; index += 1) { const arg = argv[index]; const value = argv[index + 1]; if (arg === '--config' && value) { options.configPath = path.resolve(value); index += 1; } else if (arg === '--help' || arg === '-h') { options.help = true; } } return options; } function printHelp() { console.log(`JHM TCP Socket 网关服务 用法: node app.js node app.js --config ./config.json jhm-service.exe --config .\\runtime\\config.json ./jhm-service --config ./runtime/config.json 参数: --config 指定配置文件路径 --help, -h 显示帮助 `); } function loadConfig(configPath) { if (!fs.existsSync(configPath)) { throw new Error(`未找到配置文件: ${configPath}`); } const content = fs.readFileSync(configPath, 'utf8'); const config = JSON.parse(content); validateConfig(config); return config; } function getSendChannels(config) { const channels = Array.isArray(config.send && config.send.channels) ? config.send.channels : ['mqtt']; return Array.from(new Set(channels .map((item) => String(item || '').trim().toLowerCase()) .filter(Boolean))); } function getSendOptions(config) { const send = config.send || {}; const mode = ['immediate', 'batch'].includes(String(send.mode || '').trim().toLowerCase()) ? String(send.mode).trim().toLowerCase() : 'batch'; return { mode, flushIntervalMs: Number(send.flushIntervalMs) > 0 ? Number(send.flushIntervalMs) : 60000, alignToMinute: send.alignToMinute !== false, includeDeviceIdField: send.includeDeviceIdField !== false, deviceIdField: send.deviceIdField || 'n', publishOnShutdown: send.publishOnShutdown !== false, }; } function getBloodPressureOptions(config) { const bloodPressure = config.protocol && config.protocol.bloodPressure; return { publishTime: bloodPressure ? bloodPressure.publishTime !== false : true, flushImmediately: bloodPressure ? bloodPressure.flushImmediately !== false : true, }; } function validateConfig(config) { if (!config.tcp || !Array.isArray(config.devices)) { throw new Error('config.json 必须包含 tcp 和 devices'); } if (!config.tcp.host || !config.tcp.port) { throw new Error('config.json 必须包含 tcp.host 和 tcp.port'); } const channels = getSendChannels(config); if (channels.length === 0) { throw new Error('config.json 的 send.channels 至少要启用一个通道'); } if (channels.includes('mqtt')) { if (!config.mqtt) { throw new Error('config.json 已启用 mqtt,但缺少 mqtt 配置'); } const hasBrokerUrl = Boolean(config.mqtt.brokerUrl); const hasHostMode = Boolean(config.mqtt.protocol && config.mqtt.host && config.mqtt.port); if (!hasBrokerUrl && !hasHostMode) { throw new Error('config.json 的 mqtt 必须配置 brokerUrl 或 protocol/host/port'); } if (!config.mqtt.topicTemplate && !config.mqtt.defaultTopicPrefix) { throw new Error('config.json 的 mqtt 必须配置 topicTemplate 或 defaultTopicPrefix'); } } if (channels.includes('aliyun')) { if (!config.aliyun) { throw new Error('config.json 已启用 aliyun,但缺少 aliyun 配置'); } if (!config.aliyun.tupleApiBaseUrl && !config.aliyun.tupleApiUrl) { throw new Error('config.json 的 aliyun 必须配置 tupleApiBaseUrl 或 tupleApiUrl'); } } if (!config.protocol || !config.protocol.alModelPath) { throw new Error('config.json 必须配置 protocol.alModelPath'); } const sendOptions = getSendOptions(config); if (sendOptions.flushIntervalMs <= 0) { throw new Error('config.json 的 send.flushIntervalMs 必须大于 0'); } if (config.logging && config.logging.level) { const supportedLevels = ['debug', 'info', 'warn', 'error']; if (!supportedLevels.includes(String(config.logging.level).toLowerCase())) { throw new Error(`config.json 的 logging.level 仅支持 ${supportedLevels.join('、')}`); } } } function resolveConfigPath(configFilePath, targetPath) { if (path.isAbsolute(targetPath)) { return targetPath; } return path.join(path.dirname(configFilePath), targetPath); } function createOnMetricHandler({ logger, aggregator, sendOptions, bloodPressureOptions }) { return function onMetric(device, metric, result) { logger.info(`[APP] 指标已缓存 deviceId=${device.deviceId} mode=${sendOptions.mode} metric=${JSON.stringify(metric)}`); return aggregator.ingest(device, metric) .then(() => { if ( sendOptions.mode === 'batch' && bloodPressureOptions.flushImmediately && result && result.protocol === 'blood-pressure' ) { logger.info(`[APP] 血压报文触发整包立即发送 deviceId=${device.deviceId}`); return aggregator.flush({ reason: 'blood-pressure', deviceId: device.deviceId }); } return null; }) .catch((error) => { logger.error(`[APP] 指标处理失败 deviceId=${device.deviceId}: ${error.message}`); }); }; } async function main() { const options = parseArgs(process.argv); if (options.help) { printHelp(); return; } const bootstrapLogger = createLogger({ enabled: false, console: true, }); bootstrapLogger.info(`[APP] 启动参数解析完成 config=${options.configPath}`); const config = loadConfig(options.configPath); const logDir = resolveConfigPath(options.configPath, (config.logging && config.logging.dir) || './logs'); const logger = createLogger({ ...(config.logging || {}), dir: logDir, }); const sendChannels = getSendChannels(config); const sendOptions = getSendOptions(config); const bloodPressureOptions = getBloodPressureOptions(config); const alModelPath = resolveConfigPath(options.configPath, config.protocol.alModelPath); if (config.logging && config.logging.enabled === false) { logger.info('[APP] 文件日志已禁用,仅输出到控制台'); } else { logger.info(`[APP] 本地日志已启用 dir=${logDir} file=${logger.getLogFilePath()} level=${normalizeLogLevel(config.logging && config.logging.level)}`); } logger.info(`[APP] 配置加载完成 tcp=${config.tcp.host}:${config.tcp.port} devices=${config.devices.length} channels=${sendChannels.join(',')} sendMode=${sendOptions.mode} flushIntervalMs=${sendOptions.flushIntervalMs} alModel=${alModelPath}`); const mqttService = sendChannels.includes('mqtt') ? new MqttService(config.mqtt, logger) : null; const aliyunService = sendChannels.includes('aliyun') ? new AliyunService(config.aliyun, logger) : null; // 聚合器负责把多次单点指标合并成一次完整物模型上报。 const aggregator = new MetricAggregator({ logger, ...sendOptions, onFlush: async (device, payload, meta) => { logger.info(`[APP] 开始批量分发 deviceId=${device.deviceId} channels=${sendChannels.join(',')} reason=${meta.reason} payload=${JSON.stringify(payload)}`); let ok = true; if (mqttService) { try { mqttService.publish(device, payload); } catch (error) { ok = false; logger.error(`[APP] MQTT 批量分发失败 deviceId=${device.deviceId}: ${error.message}`); } } if (aliyunService) { try { const result = await aliyunService.publish(device, payload); if (result && result.skipped) { ok = false; logger.warn(`[APP] 阿里云批量分发已跳过 deviceId=${device.deviceId} reason=${result.reason}`); } else if (result && result.ok === false) { ok = false; logger.error(`[APP] 阿里云批量分发失败 deviceId=${device.deviceId}: ${result.reason}`); } } catch (error) { ok = false; logger.error(`[APP] 阿里云批量分发异常 deviceId=${device.deviceId}: ${error.message}`); } } return ok; }, }); const tcpService = new TcpService({ tcpConfig: config.tcp, devices: config.devices, alModelPath, publishBloodPressureTime: bloodPressureOptions.publishTime, logger, onMetric: createOnMetricHandler({ logger, aggregator, sendOptions, bloodPressureOptions, }), }); if (mqttService) { logger.info('[APP] 正在启动 MQTT 通道'); mqttService.start(); } if (aliyunService) { logger.info('[APP] 正在启动阿里云通道'); aliyunService.start(); } aggregator.start(); await tcpService.start(); logger.info('[APP] 所有服务已启动'); let shuttingDown = false; async function shutdown(signal) { if (shuttingDown) { return; } shuttingDown = true; logger.warn(`[APP] 收到 ${signal},开始关闭服务`); await tcpService.stop(); await aggregator.stop(); if (mqttService) { await mqttService.stop(); } if (aliyunService) { await aliyunService.stop(); } logger.info('[APP] 服务已完成关闭'); await logger.close(); process.exit(0); } process.on('SIGINT', () => { shutdown('SIGINT'); }); process.on('SIGTERM', () => { shutdown('SIGTERM'); }); process.on('uncaughtException', (error) => { logger.error(`[APP] 未捕获异常: ${error.stack || error.message}`); }); process.on('unhandledRejection', (reason) => { logger.error(`[APP] 未处理的 Promise 拒绝: ${reason}`); }); } if (require.main === module) { main().catch((error) => { console.error(error.message); process.exit(1); }); } module.exports = { createLogger, formatLocalTimestamp, getBloodPressureOptions, getSendChannels, getSendOptions, loadConfig, main, createOnMetricHandler, normalizeLogLevel, parseArgs, printHelp, resolveConfigPath, validateConfig, };