feat: update to new cluster protocol
This commit is contained in:
@@ -70,14 +70,34 @@ udpServer.on('message', (msg, rinfo) => {
|
||||
|
||||
console.log(`📨 UDP message received from ${sourceIp}:${sourcePort}: "${message}"`);
|
||||
|
||||
if (message.startsWith('CLUSTER_HEARTBEAT:')) {
|
||||
// Handle heartbeat messages that update member list
|
||||
const hostname = message.substring('CLUSTER_HEARTBEAT:'.length);
|
||||
updateNodeFromHeartbeat(sourceIp, sourcePort, hostname);
|
||||
} else if (message.startsWith('NODE_UPDATE:')) {
|
||||
// Handle node update messages that provide detailed node info
|
||||
handleNodeUpdate(sourceIp, message);
|
||||
} else if (!message.startsWith('RAW:')) {
|
||||
// Extract topic by splitting on first ":"
|
||||
const parts = message.split(':', 2);
|
||||
if (parts.length < 2) {
|
||||
console.log(`Invalid message format from ${sourceIp}:${sourcePort}: "${message}"`);
|
||||
return;
|
||||
}
|
||||
|
||||
const topic = parts[0] + ':';
|
||||
const payload = parts[1];
|
||||
|
||||
// Handler map for different UDP message types
|
||||
const handlers = {
|
||||
'cluster/heartbeat:': (payload, sourceIp, sourcePort) => {
|
||||
updateNodeFromHeartbeat(sourceIp, sourcePort, payload);
|
||||
},
|
||||
'node/update:': (payload, sourceIp) => {
|
||||
handleNodeUpdate(sourceIp, 'node/update:' + payload);
|
||||
},
|
||||
'raw:': () => {
|
||||
// Ignore raw messages
|
||||
}
|
||||
};
|
||||
|
||||
// Look up and execute handler
|
||||
const handler = handlers[topic];
|
||||
if (handler) {
|
||||
handler(payload, sourceIp, sourcePort);
|
||||
} else {
|
||||
console.log(`Received unknown message from ${sourceIp}:${sourcePort}: "${message}"`);
|
||||
}
|
||||
} catch (error) {
|
||||
@@ -919,7 +939,7 @@ function broadcastClusterUpdate() {
|
||||
// Get cluster members asynchronously
|
||||
getCurrentClusterMembers().then(members => {
|
||||
const clusterData = {
|
||||
type: 'cluster_update',
|
||||
topic: 'cluster/update',
|
||||
members: members,
|
||||
primaryNode: primaryNodeIp,
|
||||
totalNodes: discoveredNodes.size,
|
||||
@@ -946,7 +966,7 @@ function broadcastNodeDiscovery(nodeIp, action) {
|
||||
if (wsClients.size === 0 || !wss) return;
|
||||
|
||||
const eventData = {
|
||||
type: 'node_discovery',
|
||||
topic: 'node/discovery',
|
||||
action: action, // 'discovered' or 'stale'
|
||||
nodeIp: nodeIp,
|
||||
timestamp: new Date().toISOString()
|
||||
@@ -1122,7 +1142,7 @@ const server = app.listen(PORT, '0.0.0.0', () => {
|
||||
initializeWebSocketServer(server);
|
||||
console.log('WebSocket server ready for real-time updates');
|
||||
|
||||
console.log('Waiting for CLUSTER_HEARTBEAT and NODE_UPDATE messages from SPORE nodes...');
|
||||
console.log('Waiting for cluster/heartbeat and node/update messages from SPORE nodes...');
|
||||
});
|
||||
|
||||
// Graceful shutdown handling
|
||||
|
||||
@@ -263,7 +263,8 @@ class WebSocketClient {
|
||||
try {
|
||||
const data = JSON.parse(event.data);
|
||||
logger.debug('WebSocket message received:', data);
|
||||
logger.debug('WebSocket message type:', data.type);
|
||||
const messageTopic = data.topic || data.type;
|
||||
logger.debug('WebSocket message topic:', messageTopic);
|
||||
this.emit('message', data);
|
||||
this.handleMessage(data);
|
||||
} catch (error) {
|
||||
@@ -288,21 +289,21 @@ class WebSocketClient {
|
||||
}
|
||||
|
||||
handleMessage(data) {
|
||||
switch (data.type) {
|
||||
case 'cluster_update':
|
||||
this.emit('clusterUpdate', data);
|
||||
break;
|
||||
case 'node_discovery':
|
||||
this.emit('nodeDiscovery', data);
|
||||
break;
|
||||
case 'firmware_upload_status':
|
||||
this.emit('firmwareUploadStatus', data);
|
||||
break;
|
||||
case 'rollout_progress':
|
||||
this.emit('rolloutProgress', data);
|
||||
break;
|
||||
default:
|
||||
logger.debug('Unknown WebSocket message type:', data.type);
|
||||
const messageTopic = data.topic || data.type;
|
||||
|
||||
// Handler map for different WebSocket message types
|
||||
const handlers = {
|
||||
'cluster/update': (data) => this.emit('clusterUpdate', data),
|
||||
'node/discovery': (data) => this.emit('nodeDiscovery', data),
|
||||
'firmware/upload/status': (data) => this.emit('firmwareUploadStatus', data),
|
||||
'rollout/progress': (data) => this.emit('rolloutProgress', data)
|
||||
};
|
||||
|
||||
const handler = handlers[messageTopic];
|
||||
if (handler) {
|
||||
handler(data);
|
||||
} else {
|
||||
logger.debug('Unknown WebSocket message topic:', messageTopic);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user