style: 整理代码格式
This commit is contained in:
+366
-232
@@ -1,89 +1,104 @@
|
||||
import mqtt from 'mqtt'
|
||||
import { app } from 'electron'
|
||||
import { join } from 'path'
|
||||
import { readFileSync, writeFileSync, existsSync, mkdirSync } from 'fs'
|
||||
import { app } from 'electron';
|
||||
import { readFileSync, writeFileSync, existsSync, mkdirSync } from 'fs';
|
||||
import { join } from 'path';
|
||||
|
||||
import mqtt from 'mqtt';
|
||||
|
||||
class MqttManager {
|
||||
constructor() {
|
||||
this.clients = new Map() // serverId -> mqtt client
|
||||
this.servers = new Map() // serverId -> server config
|
||||
this.messages = new Map() // serverId -> message[]
|
||||
this.publishHistory = new Map() // serverId -> publishHistory[]
|
||||
this.listeners = new Set() // IPC callback functions
|
||||
this.msgIdCounter = 0 // 用于生成唯一消息 id
|
||||
|
||||
this.dataPath = join(app.getPath('userData'), 'frost-mqtt-data.json')
|
||||
this.loadData()
|
||||
constructor() {
|
||||
this.clients = new Map(); // serverId -> mqtt client
|
||||
this.servers = new Map(); // serverId -> server config
|
||||
this.messages = new Map(); // serverId -> message[]
|
||||
this.publishHistory = new Map(); // serverId -> publishHistory[]
|
||||
this.listeners = new Set(); // IPC callback functions
|
||||
this.msgIdCounter = 0; // 用于生成唯一消息 id
|
||||
this.dataPath = join(app.getPath('userData'), 'frost-mqtt-data.json');
|
||||
this.loadData();
|
||||
}
|
||||
|
||||
nextMsgId() {
|
||||
this.msgIdCounter = (this.msgIdCounter + 1) % Number.MAX_SAFE_INTEGER
|
||||
return 'msg_' + Date.now() + '_' + this.msgIdCounter
|
||||
this.msgIdCounter = (this.msgIdCounter + 1) % Number.MAX_SAFE_INTEGER;
|
||||
return 'msg_' + Date.now() + '_' + this.msgIdCounter;
|
||||
}
|
||||
|
||||
// ==================== 数据持久化 ====================
|
||||
|
||||
getDataDir() {
|
||||
const dir = app.getPath('userData')
|
||||
|
||||
let dir = app.getPath('userData');
|
||||
|
||||
if (!existsSync(dir)) {
|
||||
mkdirSync(dir, { recursive: true })
|
||||
mkdirSync(dir, { recursive: true });
|
||||
}
|
||||
return dir
|
||||
|
||||
return dir;
|
||||
|
||||
}
|
||||
|
||||
loadData() {
|
||||
try {
|
||||
if (existsSync(this.dataPath)) {
|
||||
const raw = readFileSync(this.dataPath, 'utf-8')
|
||||
const data = JSON.parse(raw)
|
||||
|
||||
let raw = readFileSync(this.dataPath, 'utf-8');
|
||||
let data = JSON.parse(raw);
|
||||
|
||||
if (data.servers) {
|
||||
data.servers.forEach((s) => {
|
||||
s.status = 'disconnected'
|
||||
s.status = 'disconnected';
|
||||
s.topics = (s.topics || []).map((t) => ({
|
||||
...t,
|
||||
id: t.id || 'topic_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6),
|
||||
subscribed: false
|
||||
}))
|
||||
this.servers.set(s.id, s)
|
||||
})
|
||||
subscribed: false,
|
||||
}));
|
||||
this.servers.set(s.id, s);
|
||||
});
|
||||
}
|
||||
|
||||
if (data.messages) {
|
||||
Object.entries(data.messages).forEach(([id, msgs]) => {
|
||||
this.messages.set(id, msgs)
|
||||
})
|
||||
this.messages.set(id, msgs);
|
||||
});
|
||||
}
|
||||
|
||||
if (data.publishHistory) {
|
||||
Object.entries(data.publishHistory).forEach(([id, history]) => {
|
||||
this.publishHistory.set(id, history)
|
||||
})
|
||||
this.publishHistory.set(id, history);
|
||||
});
|
||||
}
|
||||
|
||||
} else {
|
||||
this.initDefaultServers()
|
||||
this.initDefaultServers();
|
||||
}
|
||||
} catch (e) {
|
||||
console.error('加载数据失败:', e.message)
|
||||
this.initDefaultServers()
|
||||
console.error('加载数据失败:', e.message);
|
||||
this.initDefaultServers();
|
||||
}
|
||||
}
|
||||
|
||||
saveData() {
|
||||
try {
|
||||
const data = {
|
||||
|
||||
let data = {
|
||||
servers: Array.from(this.servers.values()).map((s) => ({
|
||||
...s,
|
||||
status: 'disconnected',
|
||||
topics: s.topics.map((t) => ({ ...t, subscribed: false }))
|
||||
topics: s.topics.map((t) => ({ ...t, subscribed: false })),
|
||||
})),
|
||||
messages: Object.fromEntries(this.messages),
|
||||
publishHistory: Object.fromEntries(this.publishHistory)
|
||||
}
|
||||
writeFileSync(this.dataPath, JSON.stringify(data, null, 2), 'utf-8')
|
||||
publishHistory: Object.fromEntries(this.publishHistory),
|
||||
};
|
||||
|
||||
writeFileSync(this.dataPath, JSON.stringify(data, null, 2), 'utf-8');
|
||||
|
||||
} catch (e) {
|
||||
console.error('保存数据失败:', e.message)
|
||||
console.error('保存数据失败:', e.message);
|
||||
}
|
||||
}
|
||||
|
||||
initDefaultServers() {
|
||||
const defaults = [
|
||||
|
||||
let defaults = [
|
||||
{
|
||||
id: 'example_server',
|
||||
name: 'EMQX 公共测试',
|
||||
@@ -98,52 +113,58 @@ class MqttManager {
|
||||
status: 'disconnected',
|
||||
topics: [
|
||||
{ id: 'topic_default_001', topic: 'sensor/+/temperature', qos: 1, subscribed: false },
|
||||
{ id: 'topic_default_002', topic: 'device/status', qos: 2, subscribed: false }
|
||||
]
|
||||
}
|
||||
]
|
||||
{ id: 'topic_default_002', topic: 'device/status', qos: 2, subscribed: false },
|
||||
],
|
||||
},
|
||||
];
|
||||
|
||||
defaults.forEach((s) => {
|
||||
this.servers.set(s.id, s)
|
||||
this.messages.set(s.id, [])
|
||||
this.publishHistory.set(s.id, [])
|
||||
})
|
||||
this.saveData()
|
||||
this.servers.set(s.id, s);
|
||||
this.messages.set(s.id, []);
|
||||
this.publishHistory.set(s.id, []);
|
||||
});
|
||||
|
||||
this.saveData();
|
||||
|
||||
}
|
||||
|
||||
// ==================== 监听器 ====================
|
||||
|
||||
addListener(callback) {
|
||||
this.listeners.add(callback)
|
||||
this.listeners.add(callback);
|
||||
}
|
||||
|
||||
removeListener(callback) {
|
||||
this.listeners.delete(callback)
|
||||
this.listeners.delete(callback);
|
||||
}
|
||||
|
||||
notify(event, data) {
|
||||
this.listeners.forEach((cb) => {
|
||||
try {
|
||||
cb(event, data)
|
||||
cb(event, data);
|
||||
} catch (e) {
|
||||
console.error('通知失败:', e)
|
||||
console.error('通知失败:', e);
|
||||
}
|
||||
})
|
||||
});
|
||||
}
|
||||
|
||||
// ==================== 服务器管理 ====================
|
||||
|
||||
getServers() {
|
||||
return Array.from(this.servers.values()).map((s) => ({
|
||||
...s,
|
||||
password: s.password ? '******' : ''
|
||||
}))
|
||||
password: s.password ? '******' : '',
|
||||
}));
|
||||
}
|
||||
|
||||
getServer(id) {
|
||||
return this.servers.get(id) || null
|
||||
return this.servers.get(id) || null;
|
||||
}
|
||||
|
||||
addServer(config) {
|
||||
const id = 'srv_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6)
|
||||
const server = {
|
||||
|
||||
let id = 'srv_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6);
|
||||
let server = {
|
||||
id,
|
||||
name: config.name,
|
||||
host: config.host,
|
||||
@@ -155,18 +176,26 @@ class MqttManager {
|
||||
keepAlive: config.keepAlive || 60,
|
||||
cleanSession: config.cleanSession !== false,
|
||||
status: 'disconnected',
|
||||
topics: []
|
||||
}
|
||||
this.servers.set(id, server)
|
||||
this.messages.set(id, [])
|
||||
this.publishHistory.set(id, [])
|
||||
this.saveData()
|
||||
return server
|
||||
topics: [],
|
||||
};
|
||||
|
||||
this.servers.set(id, server);
|
||||
this.messages.set(id, []);
|
||||
this.publishHistory.set(id, []);
|
||||
this.saveData();
|
||||
|
||||
return server;
|
||||
|
||||
}
|
||||
|
||||
updateServer(id, config) {
|
||||
const server = this.servers.get(id)
|
||||
if (!server) return null
|
||||
|
||||
let server = this.servers.get(id);
|
||||
|
||||
if (!server) {
|
||||
return null;
|
||||
}
|
||||
|
||||
Object.assign(server, {
|
||||
name: config.name,
|
||||
host: config.host,
|
||||
@@ -175,283 +204,373 @@ class MqttManager {
|
||||
clientId: config.clientId || server.clientId,
|
||||
username: config.username || '',
|
||||
keepAlive: config.keepAlive || 60,
|
||||
cleanSession: config.cleanSession !== false
|
||||
})
|
||||
cleanSession: config.cleanSession !== false,
|
||||
});
|
||||
|
||||
if (config.password) {
|
||||
server.password = config.password
|
||||
server.password = config.password;
|
||||
}
|
||||
this.saveData()
|
||||
return server
|
||||
|
||||
this.saveData();
|
||||
|
||||
return server;
|
||||
|
||||
}
|
||||
|
||||
deleteServer(id) {
|
||||
// 先断开连接
|
||||
this.disconnect(id)
|
||||
this.servers.delete(id)
|
||||
this.messages.delete(id)
|
||||
this.publishHistory.delete(id)
|
||||
this.saveData()
|
||||
this.disconnect(id);
|
||||
this.servers.delete(id);
|
||||
this.messages.delete(id);
|
||||
this.publishHistory.delete(id);
|
||||
this.saveData();
|
||||
}
|
||||
|
||||
// ==================== 连接管理 ====================
|
||||
|
||||
connect(id) {
|
||||
const server = this.servers.get(id)
|
||||
if (!server) return false
|
||||
|
||||
let server = this.servers.get(id);
|
||||
|
||||
if (!server) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 如果已连接，先断开
|
||||
if (this.clients.has(id)) {
|
||||
this.disconnect(id)
|
||||
this.disconnect(id);
|
||||
}
|
||||
|
||||
server.status = 'connecting'
|
||||
this.notify('status-change', { id, status: 'connecting' })
|
||||
server.status = 'connecting';
|
||||
|
||||
const protocol = server.protocol || 'mqtt'
|
||||
const url = `${protocol}://${server.host}:${server.port}`
|
||||
this.notify('status-change', { id, status: 'connecting' });
|
||||
|
||||
const options = {
|
||||
let protocol = server.protocol || 'mqtt';
|
||||
let url = `${protocol}://${server.host}:${server.port}`;
|
||||
|
||||
let options = {
|
||||
clientId: server.clientId,
|
||||
username: server.username || undefined,
|
||||
password: server.password || undefined,
|
||||
keepalive: server.keepAlive,
|
||||
clean: server.cleanSession,
|
||||
reconnectPeriod: 5000,
|
||||
connectTimeout: 10000
|
||||
}
|
||||
connectTimeout: 10000,
|
||||
};
|
||||
|
||||
try {
|
||||
const client = mqtt.connect(url, options)
|
||||
|
||||
let client = mqtt.connect(url, options);
|
||||
|
||||
client.on('connect', () => {
|
||||
server.status = 'connected'
|
||||
this.notify('status-change', { id, status: 'connected' })
|
||||
server.status = 'connected';
|
||||
this.notify('status-change', { id, status: 'connected' });
|
||||
// 自动订阅已有主题
|
||||
server.topics.forEach((t) => {
|
||||
this._doSubscribe(id, t.topic, { qos: t.qos })
|
||||
})
|
||||
})
|
||||
this._doSubscribe(id, t.topic, { qos: t.qos });
|
||||
});
|
||||
});
|
||||
|
||||
client.on('error', (err) => {
|
||||
console.error(`[${server.name}] 连接错误:`, err.message)
|
||||
server.status = 'error'
|
||||
this.notify('status-change', { id, status: 'error', error: err.message })
|
||||
})
|
||||
console.error(`[${server.name}] 连接错误:`, err.message);
|
||||
server.status = 'error';
|
||||
this.notify('status-change', { id, status: 'error', error: err.message });
|
||||
});
|
||||
|
||||
client.on('close', () => {
|
||||
if (server.status === 'connected') {
|
||||
server.status = 'disconnected'
|
||||
server.status = 'disconnected';
|
||||
server.topics.forEach((t) => {
|
||||
t.subscribed = false
|
||||
})
|
||||
this.notify('status-change', { id, status: 'disconnected' })
|
||||
t.subscribed = false;
|
||||
});
|
||||
this.notify('status-change', { id, status: 'disconnected' });
|
||||
}
|
||||
})
|
||||
});
|
||||
|
||||
client.on('message', (topic, payload, packet) => {
|
||||
const msg = {
|
||||
time:
|
||||
new Date().toLocaleTimeString('zh-CN', { hour12: false }) +
|
||||
'.' +
|
||||
String(Date.now() % 1000).padStart(3, '0'),
|
||||
topic,
|
||||
|
||||
let msg = {
|
||||
time: new Date().toLocaleTimeString('zh-CN', { hour12: false }) + '.' + String(Date.now() % 1000).padStart(3, '0'),
|
||||
topic: topic,
|
||||
payload: payload.toString(),
|
||||
qos: packet.qos,
|
||||
retain: packet.retain,
|
||||
direction: 'sub',
|
||||
id: this.nextMsgId()
|
||||
}
|
||||
const msgs = this.messages.get(id) || []
|
||||
msgs.unshift(msg)
|
||||
if (msgs.length > 500) msgs.pop()
|
||||
this.messages.set(id, msgs)
|
||||
this.notify('message', { serverId: id, message: msg })
|
||||
})
|
||||
id: this.nextMsgId(),
|
||||
};
|
||||
|
||||
let msgs = this.messages.get(id) || [];
|
||||
|
||||
msgs.unshift(msg);
|
||||
|
||||
if (msgs.length > 500) {
|
||||
msgs.pop();
|
||||
}
|
||||
|
||||
this.messages.set(id, msgs);
|
||||
this.notify('message', { serverId: id, message: msg });
|
||||
|
||||
});
|
||||
|
||||
this.clients.set(id, client);
|
||||
|
||||
return true;
|
||||
|
||||
this.clients.set(id, client)
|
||||
return true
|
||||
} catch (err) {
|
||||
server.status = 'error'
|
||||
this.notify('status-change', { id, status: 'error', error: err.message })
|
||||
return false
|
||||
server.status = 'error';
|
||||
this.notify('status-change', { id, status: 'error', error: err.message });
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
disconnect(id) {
|
||||
const client = this.clients.get(id)
|
||||
|
||||
let client = this.clients.get(id);
|
||||
|
||||
if (client) {
|
||||
client.end(true)
|
||||
this.clients.delete(id)
|
||||
client.end(true);
|
||||
this.clients.delete(id);
|
||||
}
|
||||
const server = this.servers.get(id)
|
||||
|
||||
let server = this.servers.get(id);
|
||||
|
||||
if (server) {
|
||||
server.status = 'disconnected'
|
||||
server.status = 'disconnected';
|
||||
server.topics.forEach((t) => {
|
||||
t.subscribed = false
|
||||
})
|
||||
this.notify('status-change', { id, status: 'disconnected' })
|
||||
t.subscribed = false;
|
||||
});
|
||||
this.notify('status-change', { id, status: 'disconnected' });
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// ==================== 订阅管理 ====================
|
||||
_doSubscribe(serverId, topic, opts) {
|
||||
const client = this.clients.get(serverId)
|
||||
const server = this.servers.get(serverId)
|
||||
if (!client || !server || server.status !== 'connected') return false
|
||||
|
||||
let client = this.clients.get(serverId);
|
||||
let server = this.servers.get(serverId);
|
||||
|
||||
if (!client || !server || server.status !== 'connected') {
|
||||
return false;
|
||||
}
|
||||
|
||||
client.subscribe(topic, opts, (err) => {
|
||||
if (err) {
|
||||
console.error(`订阅失败 [${topic}]:`, err.message)
|
||||
this.notify('subscription-error', { serverId, topic, error: err.message })
|
||||
console.error(`订阅失败 [${topic}]:`, err.message);
|
||||
this.notify('subscription-error', { serverId, topic, error: err.message });
|
||||
} else {
|
||||
const t = server.topics.find((t) => t.topic === topic)
|
||||
if (t) t.subscribed = true
|
||||
this.notify('subscription-change', { serverId, topic, subscribed: true })
|
||||
let t = server.topics.find((t) => t.topic === topic);
|
||||
if (t) {
|
||||
t.subscribed = true;
|
||||
}
|
||||
this.notify('subscription-change', { serverId, topic, subscribed: true });
|
||||
}
|
||||
})
|
||||
return true
|
||||
});
|
||||
|
||||
return true;
|
||||
|
||||
}
|
||||
|
||||
addTopic(serverId, topic, qos = 1) {
|
||||
const server = this.servers.get(serverId)
|
||||
if (!server) return false
|
||||
if (server.topics.find((t) => t.topic === topic)) return false
|
||||
let server = this.servers.get(serverId);
|
||||
if (!server) {
|
||||
return false;
|
||||
}
|
||||
if (server.topics.find((t) => t.topic === topic)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const topicId = 'topic_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6)
|
||||
server.topics.push({ id: topicId, topic, qos, subscribed: false })
|
||||
this.saveData()
|
||||
let topicId = 'topic_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6);
|
||||
|
||||
server.topics.push({ id: topicId, topic, qos, subscribed: false });
|
||||
|
||||
this.saveData();
|
||||
|
||||
// 如果已连接，自动订阅
|
||||
if (server.status === 'connected') {
|
||||
this._doSubscribe(serverId, topic, { qos })
|
||||
this._doSubscribe(serverId, topic, { qos });
|
||||
}
|
||||
return true
|
||||
|
||||
return true;
|
||||
|
||||
}
|
||||
|
||||
removeTopic(serverId, topic) {
|
||||
const server = this.servers.get(serverId)
|
||||
if (!server) return false
|
||||
|
||||
let server = this.servers.get(serverId);
|
||||
|
||||
if (!server) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 先取消订阅
|
||||
this.unsubscribe(serverId, topic)
|
||||
server.topics = server.topics.filter((t) => t.topic !== topic)
|
||||
this.saveData()
|
||||
return true
|
||||
this.unsubscribe(serverId, topic);
|
||||
|
||||
server.topics = server.topics.filter((t) => t.topic !== topic);
|
||||
|
||||
this.saveData();
|
||||
|
||||
return true;
|
||||
|
||||
}
|
||||
|
||||
subscribe(serverId, topic) {
|
||||
const server = this.servers.get(serverId)
|
||||
if (!server) return false
|
||||
const t = server.topics.find((t) => t.topic === topic)
|
||||
if (!t) return false
|
||||
return this._doSubscribe(serverId, topic, { qos: t.qos })
|
||||
|
||||
let server = this.servers.get(serverId);
|
||||
|
||||
if (!server) {
|
||||
return false;
|
||||
}
|
||||
|
||||
let t = server.topics.find((t) => t.topic === topic);
|
||||
|
||||
if (!t) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return this._doSubscribe(serverId, topic, { qos: t.qos });
|
||||
|
||||
}
|
||||
|
||||
unsubscribe(serverId, topic) {
|
||||
const client = this.clients.get(serverId)
|
||||
const server = this.servers.get(serverId)
|
||||
if (!client || !server || server.status !== 'connected') return false
|
||||
|
||||
let client = this.clients.get(serverId);
|
||||
let server = this.servers.get(serverId);
|
||||
|
||||
if (!client || !server || server.status !== 'connected') {
|
||||
return false;
|
||||
}
|
||||
|
||||
client.unsubscribe(topic, {}, (err) => {
|
||||
if (!err) {
|
||||
const t = server.topics.find((t) => t.topic === topic)
|
||||
if (t) t.subscribed = false
|
||||
this.notify('subscription-change', { serverId, topic, subscribed: false })
|
||||
let t = server.topics.find((t) => t.topic === topic);
|
||||
if (t) {
|
||||
t.subscribed = false;
|
||||
}
|
||||
this.notify('subscription-change', { serverId, topic, subscribed: false });
|
||||
}
|
||||
})
|
||||
return true
|
||||
});
|
||||
|
||||
return true;
|
||||
|
||||
}
|
||||
|
||||
updateTopic(serverId, topicId, newTopic, newQos) {
|
||||
const server = this.servers.get(serverId)
|
||||
if (!server) return false
|
||||
const t = server.topics.find((item) => item.id === topicId)
|
||||
if (!t) return false
|
||||
|
||||
const topicChanged = t.topic !== newTopic
|
||||
let server = this.servers.get(serverId);
|
||||
|
||||
if (!server) {
|
||||
return false;
|
||||
}
|
||||
|
||||
let t = server.topics.find((item) => item.id === topicId);
|
||||
|
||||
if (!t) {
|
||||
return false;
|
||||
}
|
||||
|
||||
let topicChanged = t.topic !== newTopic;
|
||||
|
||||
if (
|
||||
topicChanged &&
|
||||
server.topics.find((item) => item.topic === newTopic && item.id !== topicId)
|
||||
) {
|
||||
return false
|
||||
return false;
|
||||
}
|
||||
|
||||
const oldTopic = t.topic
|
||||
const wasSubscribed = t.subscribed
|
||||
let oldTopic = t.topic;
|
||||
let wasSubscribed = t.subscribed;
|
||||
|
||||
// 如果主题名变更且已订阅，先取消旧主题订阅
|
||||
if (topicChanged && wasSubscribed) {
|
||||
this.unsubscribe(serverId, oldTopic)
|
||||
this.unsubscribe(serverId, oldTopic);
|
||||
}
|
||||
|
||||
t.topic = newTopic
|
||||
t.qos = newQos
|
||||
t.topic = newTopic;
|
||||
t.qos = newQos;
|
||||
|
||||
// 如果已连接且此前已订阅，重新订阅新主题
|
||||
if (server.status === 'connected' && wasSubscribed) {
|
||||
this._doSubscribe(serverId, newTopic, { qos: newQos })
|
||||
this._doSubscribe(serverId, newTopic, { qos: newQos });
|
||||
}
|
||||
|
||||
this.saveData()
|
||||
this.saveData();
|
||||
this.notify('subscription-change', {
|
||||
serverId,
|
||||
topic: newTopic,
|
||||
subscribed: t.subscribed,
|
||||
topicId
|
||||
})
|
||||
return true
|
||||
topicId,
|
||||
});
|
||||
|
||||
return true;
|
||||
|
||||
}
|
||||
|
||||
subscribeAll(serverId) {
|
||||
const client = this.clients.get(serverId)
|
||||
const server = this.servers.get(serverId)
|
||||
if (!client || !server || server.status !== 'connected') return false
|
||||
|
||||
const topicsToSubscribe = server.topics.filter((t) => !t.subscribed)
|
||||
if (topicsToSubscribe.length === 0) return true
|
||||
let client = this.clients.get(serverId);
|
||||
let server = this.servers.get(serverId);
|
||||
|
||||
const subMap = Object.fromEntries(topicsToSubscribe.map((t) => [t.topic, { qos: t.qos }]))
|
||||
if (!client || !server || server.status !== 'connected') {
|
||||
return false;
|
||||
}
|
||||
|
||||
let topicsToSubscribe = server.topics.filter((t) => !t.subscribed);
|
||||
|
||||
if (topicsToSubscribe.length === 0) {
|
||||
return true;
|
||||
}
|
||||
|
||||
let subMap = Object.fromEntries(topicsToSubscribe.map((t) => [t.topic, { qos: t.qos }]));
|
||||
|
||||
client.subscribe(subMap, (err) => {
|
||||
if (err) {
|
||||
console.error('批量订阅失败:', err.message)
|
||||
console.error('批量订阅失败:', err.message);
|
||||
topicsToSubscribe.forEach((t) => {
|
||||
this.notify('subscription-error', { serverId, topic: t.topic, error: err.message })
|
||||
})
|
||||
this.notify('subscription-error', { serverId, topic: t.topic, error: err.message });
|
||||
});
|
||||
} else {
|
||||
topicsToSubscribe.forEach((t) => {
|
||||
t.subscribed = true
|
||||
t.subscribed = true;
|
||||
this.notify('subscription-change', {
|
||||
serverId,
|
||||
topic: t.topic,
|
||||
subscribed: true,
|
||||
topicId: t.id
|
||||
})
|
||||
})
|
||||
topicId: t.id,
|
||||
});
|
||||
});
|
||||
}
|
||||
})
|
||||
return true
|
||||
});
|
||||
|
||||
return true;
|
||||
|
||||
}
|
||||
|
||||
// ==================== 消息发布 ====================
|
||||
publish(serverId, topic, payload, opts = {}) {
|
||||
const client = this.clients.get(serverId)
|
||||
const server = this.servers.get(serverId)
|
||||
if (!client || !server || server.status !== 'connected') return false
|
||||
|
||||
const pubOpts = {
|
||||
qos: opts.qos ?? 1,
|
||||
retain: opts.retain ?? false
|
||||
publish(serverId, topic, payload, opts = {}) {
|
||||
|
||||
let client = this.clients.get(serverId);
|
||||
let server = this.servers.get(serverId);
|
||||
|
||||
if (!client || !server || server.status !== 'connected') {
|
||||
return false;
|
||||
}
|
||||
|
||||
let pubOpts = {
|
||||
qos: opts.qos ?? 1,
|
||||
retain: opts.retain ?? false,
|
||||
};
|
||||
|
||||
client.publish(topic, payload, pubOpts, (err) => {
|
||||
if (err) {
|
||||
console.error(`发布失败 [${topic}]:`, err.message)
|
||||
this.notify('publish-error', { serverId, topic, error: err.message })
|
||||
console.error(`发布失败 [${topic}]:`, err.message);
|
||||
this.notify('publish-error', { serverId, topic, error: err.message });
|
||||
}
|
||||
})
|
||||
});
|
||||
|
||||
// 记录发布历史
|
||||
const pubEntry = {
|
||||
let pubEntry = {
|
||||
time:
|
||||
new Date().toLocaleTimeString('zh-CN', { hour12: false }) +
|
||||
'.' +
|
||||
@@ -460,55 +579,70 @@ class MqttManager {
|
||||
payload,
|
||||
qos: pubOpts.qos,
|
||||
retain: pubOpts.retain,
|
||||
direction: 'pub'
|
||||
direction: 'pub',
|
||||
};
|
||||
|
||||
let history = this.publishHistory.get(serverId) || [];
|
||||
|
||||
history.unshift(pubEntry);
|
||||
|
||||
if (history.length > 50) {
|
||||
history.pop();
|
||||
}
|
||||
const history = this.publishHistory.get(serverId) || []
|
||||
history.unshift(pubEntry)
|
||||
if (history.length > 50) history.pop()
|
||||
this.publishHistory.set(serverId, history)
|
||||
|
||||
this.publishHistory.set(serverId, history);
|
||||
|
||||
// 同时添加到消息日志
|
||||
const pubMsgId = this.nextMsgId()
|
||||
const msgs = this.messages.get(serverId) || []
|
||||
let pubMsgId = this.nextMsgId();
|
||||
let msgs = this.messages.get(serverId) || [];
|
||||
|
||||
msgs.unshift({
|
||||
...pubEntry,
|
||||
id: pubMsgId
|
||||
})
|
||||
if (msgs.length > 500) msgs.pop()
|
||||
this.messages.set(serverId, msgs)
|
||||
id: pubMsgId,
|
||||
});
|
||||
|
||||
if (msgs.length > 500) {
|
||||
msgs.pop();
|
||||
}
|
||||
|
||||
this.messages.set(serverId, msgs);
|
||||
|
||||
this.notify('message', {
|
||||
serverId,
|
||||
message: {
|
||||
...pubEntry,
|
||||
id: pubMsgId
|
||||
}
|
||||
})
|
||||
id: pubMsgId,
|
||||
},
|
||||
});
|
||||
|
||||
this.saveData();
|
||||
|
||||
return true;
|
||||
|
||||
this.saveData()
|
||||
return true
|
||||
}
|
||||
|
||||
// ==================== 消息管理 ====================
|
||||
|
||||
getMessages(serverId) {
|
||||
return this.messages.get(serverId) || []
|
||||
return this.messages.get(serverId) || [];
|
||||
}
|
||||
|
||||
getPublishHistory(serverId) {
|
||||
return this.publishHistory.get(serverId) || []
|
||||
return this.publishHistory.get(serverId) || [];
|
||||
}
|
||||
|
||||
clearMessages(serverId) {
|
||||
this.messages.set(serverId, [])
|
||||
this.saveData()
|
||||
this.messages.set(serverId, []);
|
||||
this.saveData();
|
||||
}
|
||||
|
||||
clearPublishHistory(serverId) {
|
||||
this.publishHistory.set(serverId, [])
|
||||
this.saveData()
|
||||
this.publishHistory.set(serverId, []);
|
||||
this.saveData();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// 单例
|
||||
const mqttManager = new MqttManager()
|
||||
export default mqttManager
|
||||
const mqttManager = new MqttManager();
|
||||
export default mqttManager;
|
||||
|
||||
Reference in New Issue
Block a user