feat(mqtt): add broker manager with connect/disconnect and message handling (Task 9)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
aaac2bedff
commit
db3ece1862
287
display/middleware/src/modules/mqtt/broker-manager.ts
Normal file
287
display/middleware/src/modules/mqtt/broker-manager.ts
Normal file
@ -0,0 +1,287 @@
|
||||
import * as mqtt from 'mqtt';
|
||||
import { getDb } from '../../db/index';
|
||||
import { extractValue } from './payload-parser';
|
||||
import {
|
||||
getDatapointByTopicId,
|
||||
updateDatapointValue,
|
||||
} from '../datapoints/datapoint-manager';
|
||||
|
||||
export interface MqttBroker {
|
||||
id: number;
|
||||
name: string;
|
||||
host: string;
|
||||
port: number;
|
||||
username: string | null;
|
||||
password: string | null;
|
||||
use_tls: number;
|
||||
enabled: number;
|
||||
status: string;
|
||||
created_at: string;
|
||||
updated_at: string;
|
||||
}
|
||||
|
||||
export interface CreateBrokerData {
|
||||
name: string;
|
||||
host: string;
|
||||
port?: number;
|
||||
username?: string;
|
||||
password?: string;
|
||||
use_tls?: number;
|
||||
enabled?: number;
|
||||
}
|
||||
|
||||
export type UpdateBrokerData = Partial<CreateBrokerData>;
|
||||
|
||||
interface MqttTopic {
|
||||
id: number;
|
||||
broker_id: number;
|
||||
topic: string;
|
||||
json_path: string | null;
|
||||
selected: number;
|
||||
ignored: number;
|
||||
}
|
||||
|
||||
const connections = new Map<number, mqtt.MqttClient>();
|
||||
|
||||
export function getAllBrokers(): MqttBroker[] {
|
||||
const db = getDb();
|
||||
return db.prepare('SELECT * FROM mqtt_brokers ORDER BY id ASC').all() as MqttBroker[];
|
||||
}
|
||||
|
||||
export function getBrokerById(id: number): MqttBroker | undefined {
|
||||
const db = getDb();
|
||||
return db.prepare('SELECT * FROM mqtt_brokers WHERE id = ?').get(id) as MqttBroker | undefined;
|
||||
}
|
||||
|
||||
export function createBroker(data: CreateBrokerData): MqttBroker {
|
||||
const db = getDb();
|
||||
const result = db.run(
|
||||
`INSERT INTO mqtt_brokers (name, host, port, username, password, use_tls, enabled)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)`,
|
||||
[
|
||||
data.name,
|
||||
data.host,
|
||||
data.port ?? 1883,
|
||||
data.username ?? null,
|
||||
data.password ?? null,
|
||||
data.use_tls ?? 0,
|
||||
data.enabled ?? 1,
|
||||
]
|
||||
);
|
||||
const id = result.lastInsertRowid as number;
|
||||
return db.prepare('SELECT * FROM mqtt_brokers WHERE id = ?').get(id) as MqttBroker;
|
||||
}
|
||||
|
||||
export function updateBroker(id: number, data: UpdateBrokerData): MqttBroker | undefined {
|
||||
const db = getDb();
|
||||
const fields: string[] = [];
|
||||
const values: unknown[] = [];
|
||||
|
||||
if (data.name !== undefined) { fields.push('name = ?'); values.push(data.name); }
|
||||
if (data.host !== undefined) { fields.push('host = ?'); values.push(data.host); }
|
||||
if (data.port !== undefined) { fields.push('port = ?'); values.push(data.port); }
|
||||
if (data.username !== undefined) { fields.push('username = ?'); values.push(data.username); }
|
||||
if (data.password !== undefined) { fields.push('password = ?'); values.push(data.password); }
|
||||
if (data.use_tls !== undefined) { fields.push('use_tls = ?'); values.push(data.use_tls); }
|
||||
if (data.enabled !== undefined) { fields.push('enabled = ?'); values.push(data.enabled); }
|
||||
|
||||
if (fields.length === 0) return getBrokerById(id);
|
||||
|
||||
fields.push("updated_at = datetime('now')");
|
||||
values.push(id);
|
||||
db.run(`UPDATE mqtt_brokers SET ${fields.join(', ')} WHERE id = ?`, values);
|
||||
return getBrokerById(id);
|
||||
}
|
||||
|
||||
export function deleteBroker(id: number): boolean {
|
||||
const db = getDb();
|
||||
disconnectBroker(id);
|
||||
const result = db.run('DELETE FROM mqtt_brokers WHERE id = ?', [id]);
|
||||
return (result.changes as number) > 0;
|
||||
}
|
||||
|
||||
function buildConnectUrl(broker: MqttBroker): string {
|
||||
const protocol = broker.use_tls ? 'mqtts' : 'mqtt';
|
||||
return `${protocol}://${broker.host}:${broker.port}`;
|
||||
}
|
||||
|
||||
function setStatus(id: number, status: string): void {
|
||||
const db = getDb();
|
||||
db.run(
|
||||
`UPDATE mqtt_brokers SET status = ?, updated_at = datetime('now') WHERE id = ?`,
|
||||
[status, id]
|
||||
);
|
||||
}
|
||||
|
||||
export function connectBroker(id: number): Promise<mqtt.MqttClient> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const broker = getBrokerById(id);
|
||||
if (!broker) {
|
||||
reject(new Error(`Broker ${id} not found`));
|
||||
return;
|
||||
}
|
||||
|
||||
// Disconnect existing connection if any
|
||||
if (connections.has(id)) {
|
||||
connections.get(id)!.end(true);
|
||||
connections.delete(id);
|
||||
}
|
||||
|
||||
const url = buildConnectUrl(broker);
|
||||
const options: mqtt.IClientOptions = {
|
||||
clientId: `display-middleware-${id}-${Date.now()}`,
|
||||
reconnectPeriod: 5000,
|
||||
connectTimeout: 10000,
|
||||
};
|
||||
|
||||
if (broker.username) options.username = broker.username;
|
||||
if (broker.password) options.password = broker.password;
|
||||
|
||||
const client = mqtt.connect(url, options);
|
||||
|
||||
client.on('connect', () => {
|
||||
setStatus(id, 'connected');
|
||||
connections.set(id, client);
|
||||
|
||||
// Subscribe to all selected topics for this broker
|
||||
const db = getDb();
|
||||
const topics = db
|
||||
.prepare(
|
||||
'SELECT * FROM mqtt_topics WHERE broker_id = ? AND selected = 1 AND ignored = 0'
|
||||
)
|
||||
.all(id) as MqttTopic[];
|
||||
|
||||
for (const t of topics) {
|
||||
client.subscribe(t.topic, (err) => {
|
||||
if (err) {
|
||||
console.error(`Failed to subscribe to ${t.topic}:`, err.message);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
resolve(client);
|
||||
});
|
||||
|
||||
client.on('message', (topic: string, payloadBuf: Buffer) => {
|
||||
const db = getDb();
|
||||
const payload = payloadBuf.toString();
|
||||
|
||||
// Find the topic record
|
||||
const topicRecord = db
|
||||
.prepare(
|
||||
'SELECT * FROM mqtt_topics WHERE broker_id = ? AND topic = ?'
|
||||
)
|
||||
.get(id, topic) as MqttTopic | undefined;
|
||||
|
||||
if (!topicRecord) return;
|
||||
|
||||
// Update last_payload
|
||||
db.run(
|
||||
'UPDATE mqtt_topics SET last_payload = ? WHERE id = ?',
|
||||
[payload, topicRecord.id]
|
||||
);
|
||||
|
||||
// Extract value and update datapoint
|
||||
const extracted = extractValue(payload, topicRecord.json_path);
|
||||
if (extracted === null) return;
|
||||
|
||||
const dp = getDatapointByTopicId(topicRecord.id);
|
||||
if (dp) {
|
||||
updateDatapointValue(dp.id, extracted);
|
||||
}
|
||||
});
|
||||
|
||||
client.on('error', (err: Error) => {
|
||||
setStatus(id, 'error');
|
||||
console.error(`MQTT broker ${id} error:`, err.message);
|
||||
});
|
||||
|
||||
client.on('offline', () => {
|
||||
setStatus(id, 'disconnected');
|
||||
});
|
||||
|
||||
client.on('reconnect', () => {
|
||||
setStatus(id, 'reconnecting');
|
||||
});
|
||||
|
||||
// Reject if connection fails within timeout
|
||||
const timer = setTimeout(() => {
|
||||
if (!connections.has(id)) {
|
||||
client.end(true);
|
||||
setStatus(id, 'error');
|
||||
reject(new Error(`Connection to broker ${id} timed out`));
|
||||
}
|
||||
}, 15000);
|
||||
|
||||
client.on('connect', () => clearTimeout(timer));
|
||||
});
|
||||
}
|
||||
|
||||
export function disconnectBroker(id: number): void {
|
||||
const client = connections.get(id);
|
||||
if (client) {
|
||||
client.end(true);
|
||||
connections.delete(id);
|
||||
setStatus(id, 'disconnected');
|
||||
}
|
||||
}
|
||||
|
||||
export function testBroker(data: {
|
||||
host: string;
|
||||
port?: number;
|
||||
username?: string;
|
||||
password?: string;
|
||||
use_tls?: number;
|
||||
}): Promise<boolean> {
|
||||
return new Promise((resolve) => {
|
||||
const protocol = data.use_tls ? 'mqtts' : 'mqtt';
|
||||
const port = data.port ?? 1883;
|
||||
const url = `${protocol}://${data.host}:${port}`;
|
||||
|
||||
const options: mqtt.IClientOptions = {
|
||||
clientId: `display-test-${Date.now()}`,
|
||||
connectTimeout: 5000,
|
||||
reconnectPeriod: 0,
|
||||
};
|
||||
if (data.username) options.username = data.username;
|
||||
if (data.password) options.password = data.password;
|
||||
|
||||
const client = mqtt.connect(url, options);
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
client.end(true);
|
||||
resolve(false);
|
||||
}, 5000);
|
||||
|
||||
client.on('connect', () => {
|
||||
clearTimeout(timer);
|
||||
client.end(true);
|
||||
resolve(true);
|
||||
});
|
||||
|
||||
client.on('error', () => {
|
||||
clearTimeout(timer);
|
||||
client.end(true);
|
||||
resolve(false);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
export async function connectAllEnabledBrokers(): Promise<void> {
|
||||
const db = getDb();
|
||||
const brokers = db
|
||||
.prepare('SELECT * FROM mqtt_brokers WHERE enabled = 1')
|
||||
.all() as MqttBroker[];
|
||||
|
||||
for (const broker of brokers) {
|
||||
try {
|
||||
await connectBroker(broker.id);
|
||||
} catch (err) {
|
||||
console.error(`Failed to connect broker ${broker.id}:`, (err as Error).message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function getConnection(id: number): mqtt.MqttClient | undefined {
|
||||
return connections.get(id);
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user