525 lines
15 KiB
JavaScript
525 lines
15 KiB
JavaScript
|
import mqtt from 'mqtt'
|
||
|
|
import { app } from 'electron'
|
||
|
|
import { join } from 'path'
|
||
|
|
import { readFileSync, writeFileSync, existsSync, mkdirSync } from 'fs'
|
||
|
|
|
||
|
|
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.dataPath = join(app.getPath('userData'), 'frost-mqtt-data.json')
|
||
|
|
this.loadData()
|
||
|
|
}
|
||
|
|
|
||
|
|
// ==================== 数据持久化 ====================
|
||
|
|
getDataDir() {
|
||
|
|
const dir = app.getPath('userData')
|
||
|
|
if (!existsSync(dir)) {
|
||
|
|
mkdirSync(dir, { recursive: true })
|
||
|
|
}
|
||
|
|
return dir
|
||
|
|
}
|
||
|
|
|
||
|
|
loadData() {
|
||
|
|
try {
|
||
|
|
if (existsSync(this.dataPath)) {
|
||
|
|
const raw = readFileSync(this.dataPath, 'utf-8')
|
||
|
|
const data = JSON.parse(raw)
|
||
|
|
if (data.servers) {
|
||
|
|
data.servers.forEach((s) => {
|
||
|
|
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)
|
||
|
|
})
|
||
|
|
}
|
||
|
|
if (data.messages) {
|
||
|
|
Object.entries(data.messages).forEach(([id, msgs]) => {
|
||
|
|
this.messages.set(id, msgs)
|
||
|
|
})
|
||
|
|
}
|
||
|
|
if (data.publishHistory) {
|
||
|
|
Object.entries(data.publishHistory).forEach(([id, history]) => {
|
||
|
|
this.publishHistory.set(id, history)
|
||
|
|
})
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
this.initDefaultServers()
|
||
|
|
}
|
||
|
|
} catch (e) {
|
||
|
|
console.error('加载数据失败:', e.message)
|
||
|
|
this.initDefaultServers()
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
saveData() {
|
||
|
|
try {
|
||
|
|
const data = {
|
||
|
|
servers: Array.from(this.servers.values()).map((s) => ({
|
||
|
|
...s,
|
||
|
|
status: 'disconnected',
|
||
|
|
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')
|
||
|
|
} catch (e) {
|
||
|
|
console.error('保存数据失败:', e.message)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
initDefaultServers() {
|
||
|
|
const defaults = [
|
||
|
|
{
|
||
|
|
id: 'srv_001',
|
||
|
|
name: 'EMQX 公共测试',
|
||
|
|
host: 'broker.emqx.io',
|
||
|
|
port: 1883,
|
||
|
|
protocol: 'mqtt',
|
||
|
|
clientId: 'frost_client_' + Math.random().toString(36).slice(2, 8),
|
||
|
|
username: '',
|
||
|
|
password: '',
|
||
|
|
keepAlive: 60,
|
||
|
|
cleanSession: true,
|
||
|
|
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: 'srv_002',
|
||
|
|
name: '本地 Mosquitto',
|
||
|
|
host: 'localhost',
|
||
|
|
port: 1883,
|
||
|
|
protocol: 'mqtt',
|
||
|
|
clientId: 'frost_local_' + Math.random().toString(36).slice(2, 8),
|
||
|
|
username: '',
|
||
|
|
password: '',
|
||
|
|
keepAlive: 60,
|
||
|
|
cleanSession: true,
|
||
|
|
status: 'disconnected',
|
||
|
|
topics: [
|
||
|
|
{ id: 'topic_default_003', topic: 'home/+/light', qos: 1, subscribed: false },
|
||
|
|
{ id: 'topic_default_004', topic: 'home/alarm', 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()
|
||
|
|
}
|
||
|
|
|
||
|
|
// ==================== 监听器 ====================
|
||
|
|
addListener(callback) {
|
||
|
|
this.listeners.add(callback)
|
||
|
|
}
|
||
|
|
|
||
|
|
removeListener(callback) {
|
||
|
|
this.listeners.delete(callback)
|
||
|
|
}
|
||
|
|
|
||
|
|
notify(event, data) {
|
||
|
|
this.listeners.forEach((cb) => {
|
||
|
|
try {
|
||
|
|
cb(event, data)
|
||
|
|
} catch (e) {
|
||
|
|
console.error('通知失败:', e)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
}
|
||
|
|
|
||
|
|
// ==================== 服务器管理 ====================
|
||
|
|
getServers() {
|
||
|
|
return Array.from(this.servers.values()).map((s) => ({
|
||
|
|
...s,
|
||
|
|
password: s.password ? '******' : ''
|
||
|
|
}))
|
||
|
|
}
|
||
|
|
|
||
|
|
getServer(id) {
|
||
|
|
return this.servers.get(id) || null
|
||
|
|
}
|
||
|
|
|
||
|
|
addServer(config) {
|
||
|
|
const id = 'srv_' + Date.now() + '_' + Math.random().toString(36).slice(2, 6)
|
||
|
|
const server = {
|
||
|
|
id,
|
||
|
|
name: config.name,
|
||
|
|
host: config.host,
|
||
|
|
port: config.port || 1883,
|
||
|
|
protocol: config.protocol || 'mqtt',
|
||
|
|
clientId: config.clientId || 'frost_' + Math.random().toString(36).slice(2, 10),
|
||
|
|
username: config.username || '',
|
||
|
|
password: config.password || '',
|
||
|
|
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
|
||
|
|
}
|
||
|
|
|
||
|
|
updateServer(id, config) {
|
||
|
|
const server = this.servers.get(id)
|
||
|
|
if (!server) return null
|
||
|
|
Object.assign(server, {
|
||
|
|
name: config.name,
|
||
|
|
host: config.host,
|
||
|
|
port: config.port || 1883,
|
||
|
|
protocol: config.protocol || 'mqtt',
|
||
|
|
clientId: config.clientId || server.clientId,
|
||
|
|
username: config.username || '',
|
||
|
|
keepAlive: config.keepAlive || 60,
|
||
|
|
cleanSession: config.cleanSession !== false
|
||
|
|
})
|
||
|
|
if (config.password) {
|
||
|
|
server.password = config.password
|
||
|
|
}
|
||
|
|
this.saveData()
|
||
|
|
return server
|
||
|
|
}
|
||
|
|
|
||
|
|
deleteServer(id) {
|
||
|
|
// 先断开连接
|
||
|
|
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
|
||
|
|
|
||
|
|
// 如果已连接，先断开
|
||
|
|
if (this.clients.has(id)) {
|
||
|
|
this.disconnect(id)
|
||
|
|
}
|
||
|
|
|
||
|
|
server.status = 'connecting'
|
||
|
|
this.notify('status-change', { id, status: 'connecting' })
|
||
|
|
|
||
|
|
const protocol = server.protocol || 'mqtt'
|
||
|
|
const url = `${protocol}://${server.host}:${server.port}`
|
||
|
|
|
||
|
|
const options = {
|
||
|
|
clientId: server.clientId,
|
||
|
|
username: server.username || undefined,
|
||
|
|
password: server.password || undefined,
|
||
|
|
keepalive: server.keepAlive,
|
||
|
|
clean: server.cleanSession,
|
||
|
|
reconnectPeriod: 5000,
|
||
|
|
connectTimeout: 10000
|
||
|
|
}
|
||
|
|
|
||
|
|
try {
|
||
|
|
const client = mqtt.connect(url, options)
|
||
|
|
|
||
|
|
client.on('connect', () => {
|
||
|
|
server.status = 'connected'
|
||
|
|
this.notify('status-change', { id, status: 'connected' })
|
||
|
|
// 自动订阅已有主题
|
||
|
|
server.topics.forEach((t) => {
|
||
|
|
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 })
|
||
|
|
})
|
||
|
|
|
||
|
|
client.on('close', () => {
|
||
|
|
if (server.status === 'connected') {
|
||
|
|
server.status = 'disconnected'
|
||
|
|
server.topics.forEach((t) => {
|
||
|
|
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,
|
||
|
|
payload: payload.toString(),
|
||
|
|
qos: packet.qos,
|
||
|
|
retain: packet.retain,
|
||
|
|
direction: 'sub',
|
||
|
|
id: 'msg_' + Date.now()
|
||
|
|
}
|
||
|
|
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 })
|
||
|
|
})
|
||
|
|
|
||
|
|
this.clients.set(id, client)
|
||
|
|
return true
|
||
|
|
} catch (err) {
|
||
|
|
server.status = 'error'
|
||
|
|
this.notify('status-change', { id, status: 'error', error: err.message })
|
||
|
|
return false
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
disconnect(id) {
|
||
|
|
const client = this.clients.get(id)
|
||
|
|
if (client) {
|
||
|
|
client.end(true)
|
||
|
|
this.clients.delete(id)
|
||
|
|
}
|
||
|
|
const server = this.servers.get(id)
|
||
|
|
if (server) {
|
||
|
|
server.status = 'disconnected'
|
||
|
|
server.topics.forEach((t) => {
|
||
|
|
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
|
||
|
|
|
||
|
|
client.subscribe(topic, opts, (err) => {
|
||
|
|
if (err) {
|
||
|
|
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 })
|
||
|
|
}
|
||
|
|
})
|
||
|
|
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
|
||
|
|
|
||
|
|
const 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 })
|
||
|
|
}
|
||
|
|
return true
|
||
|
|
}
|
||
|
|
|
||
|
|
removeTopic(serverId, topic) {
|
||
|
|
const 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
|
||
|
|
}
|
||
|
|
|
||
|
|
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 })
|
||
|
|
}
|
||
|
|
|
||
|
|
unsubscribe(serverId, topic) {
|
||
|
|
const client = this.clients.get(serverId)
|
||
|
|
const 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 })
|
||
|
|
}
|
||
|
|
})
|
||
|
|
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
|
||
|
|
if (
|
||
|
|
topicChanged &&
|
||
|
|
server.topics.find((item) => item.topic === newTopic && item.id !== topicId)
|
||
|
|
) {
|
||
|
|
return false
|
||
|
|
}
|
||
|
|
|
||
|
|
const oldTopic = t.topic
|
||
|
|
const wasSubscribed = t.subscribed
|
||
|
|
|
||
|
|
// 如果主题名变更且已订阅，先取消旧主题订阅
|
||
|
|
if (topicChanged && wasSubscribed) {
|
||
|
|
this.unsubscribe(serverId, oldTopic)
|
||
|
|
}
|
||
|
|
|
||
|
|
t.topic = newTopic
|
||
|
|
t.qos = newQos
|
||
|
|
|
||
|
|
// 如果已连接且此前已订阅，重新订阅新主题
|
||
|
|
if (server.status === 'connected' && wasSubscribed) {
|
||
|
|
this._doSubscribe(serverId, newTopic, { qos: newQos })
|
||
|
|
}
|
||
|
|
|
||
|
|
this.saveData()
|
||
|
|
this.notify('subscription-change', {
|
||
|
|
serverId,
|
||
|
|
topic: newTopic,
|
||
|
|
subscribed: t.subscribed,
|
||
|
|
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
|
||
|
|
|
||
|
|
const subMap = Object.fromEntries(topicsToSubscribe.map((t) => [t.topic, { qos: t.qos }]))
|
||
|
|
|
||
|
|
client.subscribe(subMap, (err) => {
|
||
|
|
if (err) {
|
||
|
|
console.error('批量订阅失败:', err.message)
|
||
|
|
topicsToSubscribe.forEach((t) => {
|
||
|
|
this.notify('subscription-error', { serverId, topic: t.topic, error: err.message })
|
||
|
|
})
|
||
|
|
} else {
|
||
|
|
topicsToSubscribe.forEach((t) => {
|
||
|
|
t.subscribed = true
|
||
|
|
this.notify('subscription-change', {
|
||
|
|
serverId,
|
||
|
|
topic: t.topic,
|
||
|
|
subscribed: true,
|
||
|
|
topicId: t.id
|
||
|
|
})
|
||
|
|
})
|
||
|
|
}
|
||
|
|
})
|
||
|
|
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
|
||
|
|
}
|
||
|
|
|
||
|
|
client.publish(topic, payload, pubOpts, (err) => {
|
||
|
|
if (err) {
|
||
|
|
console.error(`发布失败 [${topic}]:`, err.message)
|
||
|
|
this.notify('publish-error', { serverId, topic, error: err.message })
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
// 记录发布历史
|
||
|
|
const pubEntry = {
|
||
|
|
time:
|
||
|
|
new Date().toLocaleTimeString('zh-CN', { hour12: false }) +
|
||
|
|
'.' +
|
||
|
|
String(Date.now() % 1000).padStart(3, '0'),
|
||
|
|
topic,
|
||
|
|
payload,
|
||
|
|
qos: pubOpts.qos,
|
||
|
|
retain: pubOpts.retain,
|
||
|
|
direction: 'pub'
|
||
|
|
}
|
||
|
|
const history = this.publishHistory.get(serverId) || []
|
||
|
|
history.unshift(pubEntry)
|
||
|
|
if (history.length > 50) history.pop()
|
||
|
|
this.publishHistory.set(serverId, history)
|
||
|
|
|
||
|
|
// 同时添加到消息日志
|
||
|
|
const msgs = this.messages.get(serverId) || []
|
||
|
|
msgs.unshift({
|
||
|
|
...pubEntry,
|
||
|
|
id: 'msg_' + Date.now()
|
||
|
|
})
|
||
|
|
if (msgs.length > 500) msgs.pop()
|
||
|
|
this.messages.set(serverId, msgs)
|
||
|
|
|
||
|
|
this.notify('message', {
|
||
|
|
serverId,
|
||
|
|
message: {
|
||
|
|
...pubEntry,
|
||
|
|
id: 'msg_' + Date.now()
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
this.saveData()
|
||
|
|
return true
|
||
|
|
}
|
||
|
|
|
||
|
|
// ==================== 消息管理 ====================
|
||
|
|
getMessages(serverId) {
|
||
|
|
return this.messages.get(serverId) || []
|
||
|
|
}
|
||
|
|
|
||
|
|
getPublishHistory(serverId) {
|
||
|
|
return this.publishHistory.get(serverId) || []
|
||
|
|
}
|
||
|
|
|
||
|
|
clearMessages(serverId) {
|
||
|
|
this.messages.set(serverId, [])
|
||
|
|
this.saveData()
|
||
|
|
}
|
||
|
|
|
||
|
|
clearPublishHistory(serverId) {
|
||
|
|
this.publishHistory.set(serverId, [])
|
||
|
|
this.saveData()
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// 单例
|
||
|
|
const mqttManager = new MqttManager()
|
||
|
|
export default mqttManager
|