428 lines
12 KiB
TypeScript
428 lines
12 KiB
TypeScript
|
import mqtt, { MqttClient, IClientOptions } from "npm:mqtt";
|
|||
|
|
|
|||
|
|
// ==== 消息／配置类型 ====
|
|||
|
|
|
|||
|
|
/** init 消息接口，从主线程传入 */
|
|||
|
|
interface InitMessage {
|
|||
|
|
type: "init";
|
|||
|
|
mqtt: {
|
|||
|
|
brokerUrl: string;
|
|||
|
|
clientId: string;
|
|||
|
|
username?: string;
|
|||
|
|
password?: string;
|
|||
|
|
reconnectInterval?: number;
|
|||
|
|
qos?: 0 | 1 | 2;
|
|||
|
|
};
|
|||
|
|
/**
|
|||
|
|
* 这个数组里每项配置一对 topic 和对应的映射规则
|
|||
|
|
*/
|
|||
|
|
mappings: Array<{
|
|||
|
|
/** MQTT 订阅的源 topic，可以包含 + 或 # 通配符 */
|
|||
|
|
sourceTopic: string;
|
|||
|
|
/** 发布时使用的目标 topic 模板，可包含 {agvId} 等占位符 */
|
|||
|
|
targetTopicTemplate: string;
|
|||
|
|
/** transformGeneric 用到的映射规则 */
|
|||
|
|
mapping: Record<string, MappingRule>;
|
|||
|
|
}>;
|
|||
|
|
instanceId: string;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** 停止 Worker 的消息 */
|
|||
|
|
interface StopMessage {
|
|||
|
|
type: "stop" | "shutdown"; // 支持 stop 和 shutdown
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// transformGeneric 中用到的规则
|
|||
|
|
type MappingRule =
|
|||
|
|
| string // 简单字符串映射，直接从源路径复制
|
|||
|
|
| {
|
|||
|
|
op: "const";
|
|||
|
|
value: any;
|
|||
|
|
source?: string;
|
|||
|
|
}
|
|||
|
|
| {
|
|||
|
|
type: "object";
|
|||
|
|
op?: "object";
|
|||
|
|
source?: string;
|
|||
|
|
mapping: Record<string, MappingRule>;
|
|||
|
|
}
|
|||
|
|
| {
|
|||
|
|
type: "array";
|
|||
|
|
op?: "array";
|
|||
|
|
source: string;
|
|||
|
|
mapping: Record<string, MappingRule>;
|
|||
|
|
}
|
|||
|
|
| {
|
|||
|
|
op: "toString";
|
|||
|
|
source: string;
|
|||
|
|
}
|
|||
|
|
| {
|
|||
|
|
// 通用对象映射（没有 type 字段）
|
|||
|
|
mapping: Record<string, MappingRule>;
|
|||
|
|
source?: string;
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
// ==== 通用工具函数 ====
|
|||
|
|
|
|||
|
|
/** 按路径 a.b.c 从 obj 取值 */
|
|||
|
|
function getByPath(obj: any, path: string): any {
|
|||
|
|
return path.split(".").reduce((o, key) => (o != null ? o[key] : undefined), obj);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** 按路径 a.b.c 在 target 上写入 value */
|
|||
|
|
function setByPath(target: any, path: string, value: any) {
|
|||
|
|
const keys = path.split(".");
|
|||
|
|
let cur = target;
|
|||
|
|
for (let i = 0; i < keys.length - 1; i++) {
|
|||
|
|
const k = keys[i];
|
|||
|
|
if (!(k in cur) || typeof cur[k] !== "object") {
|
|||
|
|
cur[k] = {};
|
|||
|
|
}
|
|||
|
|
cur = cur[k];
|
|||
|
|
}
|
|||
|
|
cur[keys[keys.length - 1]] = value;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** 通用的 JSON 转换器 */
|
|||
|
|
function transformGeneric(raw: any, mapping: Record<string, MappingRule>): any {
|
|||
|
|
const out: any = {};
|
|||
|
|
|
|||
|
|
for (const [outKey, rule] of Object.entries(mapping)) {
|
|||
|
|
try {
|
|||
|
|
// 处理简单字符串映射
|
|||
|
|
if (typeof rule === "string") {
|
|||
|
|
const value = getByPath(raw, rule);
|
|||
|
|
if (value !== undefined) {
|
|||
|
|
setByPath(out, outKey, value);
|
|||
|
|
}
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 处理对象规则
|
|||
|
|
if (typeof rule === "object" && rule !== null) {
|
|||
|
|
// 常量值映射
|
|||
|
|
if ("op" in rule && rule.op === "const") {
|
|||
|
|
setByPath(out, outKey, rule.value);
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// toString 操作
|
|||
|
|
if ("op" in rule && rule.op === "toString") {
|
|||
|
|
const sourceValue = getByPath(raw, rule.source);
|
|||
|
|
if (sourceValue !== undefined) {
|
|||
|
|
// 如果已经是字符串，直接使用；否则进行 JSON 序列化
|
|||
|
|
const stringValue = typeof sourceValue === "string"
|
|||
|
|
? sourceValue
|
|||
|
|
: JSON.stringify(sourceValue);
|
|||
|
|
setByPath(out, outKey, stringValue);
|
|||
|
|
}
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 对象映射
|
|||
|
|
if ("type" in rule && rule.type === "object") {
|
|||
|
|
const subRaw = rule.source ? getByPath(raw, rule.source) : raw;
|
|||
|
|
if (subRaw !== undefined) {
|
|||
|
|
const transformedObj = transformGeneric(subRaw, rule.mapping);
|
|||
|
|
setByPath(out, outKey, transformedObj);
|
|||
|
|
}
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 数组映射
|
|||
|
|
if ("type" in rule && rule.type === "array") {
|
|||
|
|
const sourceArray = getByPath(raw, rule.source);
|
|||
|
|
if (Array.isArray(sourceArray)) {
|
|||
|
|
const transformedArray = sourceArray.map(item =>
|
|||
|
|
transformGeneric(item, rule.mapping)
|
|||
|
|
);
|
|||
|
|
setByPath(out, outKey, transformedArray);
|
|||
|
|
} else {
|
|||
|
|
setByPath(out, outKey, []);
|
|||
|
|
}
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 如果是对象但没有特殊操作，尝试作为嵌套映射处理
|
|||
|
|
if ("mapping" in rule && typeof rule.mapping === "object") {
|
|||
|
|
const mappingRule = rule as { mapping: Record<string, MappingRule>; source?: string };
|
|||
|
|
const subRaw = mappingRule.source ? getByPath(raw, mappingRule.source) : raw;
|
|||
|
|
if (subRaw !== undefined) {
|
|||
|
|
const transformedObj = transformGeneric(subRaw, mappingRule.mapping);
|
|||
|
|
setByPath(out, outKey, transformedObj);
|
|||
|
|
}
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
console.warn(`Unknown mapping rule for key "${outKey}":`, rule);
|
|||
|
|
} catch (error) {
|
|||
|
|
console.error(`Error processing mapping rule for key "${outKey}":`, error);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return out;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/**
|
|||
|
|
* 判断一个 topic 是否匹配源 topic 模式（含 +, #）
|
|||
|
|
*/
|
|||
|
|
function topicMatches(pattern: string, topic: string): boolean {
|
|||
|
|
const patSeg = pattern.split("/");
|
|||
|
|
const topSeg = topic.split("/");
|
|||
|
|
for (let i = 0; i < patSeg.length; i++) {
|
|||
|
|
const p = patSeg[i];
|
|||
|
|
const t = topSeg[i];
|
|||
|
|
if (p === "#") return true; // 后面的全部都匹配
|
|||
|
|
if (p === "+") continue; // 单层通配符
|
|||
|
|
if (t === undefined) return false;
|
|||
|
|
if (p !== t) return false;
|
|||
|
|
}
|
|||
|
|
return patSeg.length === topSeg.length;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/**
|
|||
|
|
* 从 topic 根据 pattern 抽取 '+' 通配符对应的位置上的参数
|
|||
|
|
* 目前只支持单个 {agvId} 占位
|
|||
|
|
*/
|
|||
|
|
function extractAgvId(pattern: string, topic: string): string | null {
|
|||
|
|
const patSeg = pattern.split("/");
|
|||
|
|
const topSeg = topic.split("/");
|
|||
|
|
for (let i = 0; i < patSeg.length; i++) {
|
|||
|
|
if (patSeg[i] === "+") {
|
|||
|
|
return topSeg[i];
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
return null;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ==== 全局状态 ====
|
|||
|
|
|
|||
|
|
let mqttClient: MqttClient | null = null;
|
|||
|
|
let reconnectTimer: any = null;
|
|||
|
|
const CONFIG: {
|
|||
|
|
mqtt: {
|
|||
|
|
brokerUrl: string;
|
|||
|
|
clientId: string;
|
|||
|
|
username?: string;
|
|||
|
|
password?: string;
|
|||
|
|
reconnectInterval?: number;
|
|||
|
|
qos?: 0 | 1 | 2;
|
|||
|
|
};
|
|||
|
|
mappings: Array<{
|
|||
|
|
sourceTopic: string;
|
|||
|
|
targetTopicTemplate: string;
|
|||
|
|
mapping: Record<string, MappingRule>;
|
|||
|
|
}>;
|
|||
|
|
} = {
|
|||
|
|
mqtt: {
|
|||
|
|
brokerUrl: "",
|
|||
|
|
clientId: "",
|
|||
|
|
username: undefined,
|
|||
|
|
password: undefined,
|
|||
|
|
reconnectInterval: 5000,
|
|||
|
|
qos: 1,
|
|||
|
|
},
|
|||
|
|
mappings: [],
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
// instance identifier generated from config.manufacturer and UUID
|
|||
|
|
let INSTANCE_ID: string = "";
|
|||
|
|
|
|||
|
|
// 连续 MQTT 错误计数，只有超过10次才触发重连
|
|||
|
|
let consecutiveErrors = 0;
|
|||
|
|
|
|||
|
|
// ==== MQTT 逻辑 ====
|
|||
|
|
|
|||
|
|
async function connectAndSubscribe() {
|
|||
|
|
try {
|
|||
|
|
const opts: IClientOptions = {
|
|||
|
|
clientId: CONFIG.mqtt.clientId,
|
|||
|
|
username: CONFIG.mqtt.username,
|
|||
|
|
password: CONFIG.mqtt.password,
|
|||
|
|
reconnectPeriod: 3000, // 我们自己重连
|
|||
|
|
};
|
|||
|
|
mqttClient = mqtt.connect(CONFIG.mqtt.brokerUrl, opts);
|
|||
|
|
|
|||
|
|
// 预先注册错误、断开和关闭处理器
|
|||
|
|
mqttClient.on("error", onMqttError);
|
|||
|
|
mqttClient.on("disconnect", onMqttDisconnect);
|
|||
|
|
mqttClient.on("close", onMqttClose);
|
|||
|
|
|
|||
|
|
mqttClient.on("connect", async () => {
|
|||
|
|
// 连接成功后重置错误计数
|
|||
|
|
consecutiveErrors = 0;
|
|||
|
|
// 订阅所有 mappings 里配置的源 topic
|
|||
|
|
for (const m of CONFIG.mappings) {
|
|||
|
|
const inTopic = m.sourceTopic
|
|||
|
|
.replace("{instanceId}", INSTANCE_ID);
|
|||
|
|
// console.log("inTopic", inTopic);
|
|||
|
|
try {
|
|||
|
|
await mqttClient!.subscribe(inTopic, { qos: CONFIG.mqtt.qos! });
|
|||
|
|
} catch (err) {
|
|||
|
|
console.error(`✗ MQTT subscribe failed for ${inTopic}:`, err);
|
|||
|
|
self.postMessage({ type: "error", error: "subscribe_failed", details: (err as Error).message });
|
|||
|
|
scheduleReconnect();
|
|||
|
|
return; // 订阅失败，中断后续处理，尝试重连
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
// 订阅完成后监听消息
|
|||
|
|
mqttClient!.on("message", onMqttMessage);
|
|||
|
|
|
|||
|
|
if (reconnectTimer) {
|
|||
|
|
clearTimeout(reconnectTimer);
|
|||
|
|
reconnectTimer = null;
|
|||
|
|
}
|
|||
|
|
self.postMessage({ type: "status", status: "connected" });
|
|||
|
|
});
|
|||
|
|
} catch (err) {
|
|||
|
|
console.error("✗ MQTT connect failed:", err);
|
|||
|
|
self.postMessage({
|
|||
|
|
type: "error",
|
|||
|
|
error: "connect_failed",
|
|||
|
|
details: err instanceof Error ? err.message : String(err),
|
|||
|
|
});
|
|||
|
|
scheduleReconnect();
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async function onMqttMessage(topic: string, payload: Uint8Array) {
|
|||
|
|
try {
|
|||
|
|
const raw = JSON.parse(new TextDecoder().decode(payload));
|
|||
|
|
|
|||
|
|
// 对每个匹配的 mappingConfig，做一次转换并发布
|
|||
|
|
for (const m of CONFIG.mappings) {
|
|||
|
|
const inTopic = m.sourceTopic
|
|||
|
|
.replace("{instanceId}", INSTANCE_ID);
|
|||
|
|
if (!topicMatches(inTopic, topic)) continue;
|
|||
|
|
|
|||
|
|
// 提取占位符 agvId
|
|||
|
|
const agvId = extractAgvId(inTopic, topic) ?? "";
|
|||
|
|
|
|||
|
|
// 转换
|
|||
|
|
const transformed = transformGeneric(raw, m.mapping);
|
|||
|
|
|
|||
|
|
// 填充目标 topic
|
|||
|
|
const outTopic = m.targetTopicTemplate
|
|||
|
|
.replace("{instanceId}", INSTANCE_ID)
|
|||
|
|
.replace("{agvId}", agvId);
|
|||
|
|
|
|||
|
|
// 发布
|
|||
|
|
// console.log("outTopic", outTopic);
|
|||
|
|
await mqttClient!.publish(outTopic, JSON.stringify(transformed), {
|
|||
|
|
qos: CONFIG.mqtt.qos!,
|
|||
|
|
});
|
|||
|
|
|
|||
|
|
self.postMessage({
|
|||
|
|
type: "published",
|
|||
|
|
topic: outTopic,
|
|||
|
|
agvId,
|
|||
|
|
sourceTopic: topic,
|
|||
|
|
});
|
|||
|
|
}
|
|||
|
|
} catch (err) {
|
|||
|
|
console.error("✗ onMqttMessage error:", err);
|
|||
|
|
self.postMessage({
|
|||
|
|
type: "error",
|
|||
|
|
error: "process_message",
|
|||
|
|
details: err instanceof Error ? err.message : String(err),
|
|||
|
|
});
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
function onMqttClose() {
|
|||
|
|
console.log("vda onMqttClose");
|
|||
|
|
self.postMessage({ type: "status", status: "closed" });
|
|||
|
|
// 确保在 close 时释放底层 socket
|
|||
|
|
if (mqttClient) {
|
|||
|
|
mqttClient.end(true);
|
|||
|
|
mqttClient = null;
|
|||
|
|
}
|
|||
|
|
self.postMessage({ type: "reconnect-down" });
|
|||
|
|
// connectAndSubscribe();
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
function onMqttDisconnect() {
|
|||
|
|
self.postMessage({ type: "status", status: "disconnected" });
|
|||
|
|
console.log("vda onMqttDisconnect");
|
|||
|
|
// 断开时也关闭客户端，避免累积未关闭的连接
|
|||
|
|
if (mqttClient) {
|
|||
|
|
mqttClient.end(true);
|
|||
|
|
mqttClient = null;
|
|||
|
|
}
|
|||
|
|
scheduleReconnect();
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
function onMqttError(err: Error) {
|
|||
|
|
// 如果是连接被拒绝错误，通知主线程重启所有 worker
|
|||
|
|
const anyErr = err as any;
|
|||
|
|
if (anyErr.code === 'ECONNREFUSED' || anyErr.code === 'ECONNRESET') {
|
|||
|
|
console.error('❌ MQTT connection refused, requesting full restart:', anyErr);
|
|||
|
|
self.postMessage({ type: 'reconnect-all' });
|
|||
|
|
}
|
|||
|
|
// 增加错误计数，只在超过10次后才重连
|
|||
|
|
consecutiveErrors++;
|
|||
|
|
console.error(`! MQTT error (#${consecutiveErrors}):`, err);
|
|||
|
|
self.postMessage({
|
|||
|
|
type: "error",
|
|||
|
|
error: "mqtt_error",
|
|||
|
|
details: err.message,
|
|||
|
|
});
|
|||
|
|
if (consecutiveErrors >= 5) {
|
|||
|
|
// 重置错误计数
|
|||
|
|
consecutiveErrors = 0;
|
|||
|
|
if (mqttClient) {
|
|||
|
|
mqttClient.end(true);
|
|||
|
|
mqttClient = null;
|
|||
|
|
}
|
|||
|
|
scheduleReconnect();
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
function scheduleReconnect() {
|
|||
|
|
if (!reconnectTimer) {
|
|||
|
|
const t = CONFIG.mqtt.reconnectInterval;
|
|||
|
|
|
|||
|
|
reconnectTimer = setTimeout(() => {
|
|||
|
|
reconnectTimer = null;
|
|||
|
|
console.log("vda schedule Worker 自身重连");
|
|||
|
|
self.postMessage({ type: "reconnect-down" });
|
|||
|
|
connectAndSubscribe();
|
|||
|
|
}, t);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ==== 主线程消息处理 ====
|
|||
|
|
|
|||
|
|
self.onmessage = (e: MessageEvent<InitMessage | StopMessage>) => {
|
|||
|
|
const msg = e.data;
|
|||
|
|
if (msg.type === "init") {
|
|||
|
|
// 覆盖配置
|
|||
|
|
CONFIG.mqtt = {
|
|||
|
|
...CONFIG.mqtt,
|
|||
|
|
brokerUrl: msg.mqtt.brokerUrl,
|
|||
|
|
clientId: msg.mqtt.clientId,
|
|||
|
|
username: msg.mqtt.username,
|
|||
|
|
password: msg.mqtt.password,
|
|||
|
|
reconnectInterval: msg.mqtt.reconnectInterval ?? CONFIG.mqtt.reconnectInterval,
|
|||
|
|
qos: msg.mqtt.qos ?? CONFIG.mqtt.qos,
|
|||
|
|
};
|
|||
|
|
CONFIG.mappings = msg.mappings;
|
|||
|
|
// set our instance identifier
|
|||
|
|
INSTANCE_ID = msg.instanceId;
|
|||
|
|
console.log("vda5050_transformer_worker: instanceId", INSTANCE_ID);
|
|||
|
|
// 启动连接
|
|||
|
|
connectAndSubscribe();
|
|||
|
|
} else if (msg.type === "stop" || msg.type === "shutdown") {
|
|||
|
|
// 支持 stop 和 shutdown，统一关闭
|
|||
|
|
if (mqttClient) {
|
|||
|
|
mqttClient.end(true);
|
|||
|
|
mqttClient = null;
|
|||
|
|
}
|
|||
|
|
if (reconnectTimer) {
|
|||
|
|
clearTimeout(reconnectTimer);
|
|||
|
|
reconnectTimer = null;
|
|||
|
|
}
|
|||
|
|
self.postMessage({ type: "status", status: "stopped" });
|
|||
|
|
}
|
|||
|
|
};
|