110 lines
3.7 KiB
JavaScript
110 lines
3.7 KiB
JavaScript
const WebSocket = require('ws');
|
|
const { encode, decode } = require('@msgpack/msgpack');
|
|
const url = require('url');
|
|
const tokenStore = require('./helper/tokenStore');
|
|
const redisHelper = require('./helper/redis');
|
|
const influxWriter = require('./helper/influxWriter');
|
|
|
|
const ws = new WebSocket.Server({
|
|
port: process.env.SOCKET_PORT,
|
|
perMessageDeflate: false,
|
|
verifyClient: async (info, callback) => {
|
|
const { query } = url.parse(info.req.url, true);
|
|
const token = query.token;
|
|
|
|
if (!token) {
|
|
return callback(false, 401, 'token not passed');
|
|
}
|
|
|
|
try {
|
|
const sessionData = await tokenStore.consumeToken(token);
|
|
if (!sessionData) {
|
|
return callback(false, 401, 'token not valid or expired');
|
|
}
|
|
|
|
info.req.sensorSession = sessionData;
|
|
callback(true);
|
|
} catch (error) {
|
|
callback(false, 500, `internal server error: ${error}`);
|
|
}
|
|
}
|
|
});
|
|
|
|
ws.on('connection', async (client, req) => {
|
|
const session = req.sensorSession;
|
|
const sensorId = session.sensorId;
|
|
|
|
client.sensorId = sensorId;
|
|
|
|
// Registra la sessione su Redis
|
|
try {
|
|
await redisHelper.setSession(sensorId, {
|
|
...session.metadata,
|
|
connectedAt: Math.floor(Date.now() / 1000)
|
|
});
|
|
} catch (err) {
|
|
console.error(`[WS] Redis setSession error for ${sensorId}:`, err);
|
|
}
|
|
|
|
/**
|
|
* Gestione messaggi in arrivo dal sensore.
|
|
*
|
|
* Il sensore invia dati codificati in MessagePack (binario).
|
|
* Il formato atteso è un array compatto (per risparmiare spazio):
|
|
*
|
|
* [timestamp, measurement, { field1: value1, field2: value2, ... }]
|
|
*
|
|
* Esempio pratico (dati meteo da barca):
|
|
* [1710681000, "weather", { t: 22.5, h: 65, p: 1013.2, w: 12.3 }]
|
|
*
|
|
* Dove le chiavi abbreviate sono:
|
|
* t = temperature, h = humidity, p = pressure, w = windSpeed ...
|
|
*
|
|
* Il server decodifica il messaggio e prepara il punto per InfluxDB.
|
|
*/
|
|
client.on('message', async (raw) => {
|
|
try {
|
|
const msg = decode(raw);
|
|
|
|
// Rispondi con ACK minimale in msgpack: { a: 1 } = acknowledged
|
|
client.send(encode({ a: 1 }));
|
|
|
|
const [timestamp, measurement, fields] = msg;
|
|
|
|
// Pubblica su Redis per i watchers live
|
|
redisHelper.publishSensorData(sensorId, { timestamp, measurement, fields });
|
|
|
|
// Controlla se ci sono watchers attivi
|
|
const watchers = await redisHelper.getWatcherCount(sensorId);
|
|
|
|
if (measurement === 'forecast_batch') {
|
|
influxWriter.writeForecastBatch(sensorId, fields);
|
|
} else if (watchers > 0) {
|
|
// Con watchers: accumula nel buffer, flusha ogni 10 punti
|
|
influxWriter.bufferPoint(sensorId, timestamp, measurement, fields);
|
|
} else {
|
|
// Senza watchers: scrivi immediatamente (comportamento originale)
|
|
influxWriter.writePoint(sensorId, timestamp, measurement, fields);
|
|
}
|
|
|
|
} catch (err) {
|
|
console.error(`[WS|${sensorId}] decode error:`, err);
|
|
// Rispondi con errore in msgpack: { e: 1 } = error
|
|
client.send(encode({ e: 1 }));
|
|
}
|
|
});
|
|
|
|
client.on('error', (err) => {
|
|
console.error(`[WS|${sensorId}] error:`, err);
|
|
});
|
|
|
|
client.on('close', async () => {
|
|
try {
|
|
await redisHelper.deleteSession(sensorId);
|
|
} catch (err) {
|
|
console.error(`[WS] Redis deleteSession error for ${sensorId}:`, err);
|
|
}
|
|
});
|
|
});
|
|
|
|
console.log(`[WS] Realtime websocket server on port: ${process.env.SOCKET_PORT}`); |