Nail/tf-iot-consumer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: resilient store — Redis/Mongo errors don't stop consumer

Browse Source
This commit is contained in:
naeel
2026-07-22 09:05:26 +04:00
parent 676ed97538
commit 9b5201332e
1 changed files with 25 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.js
+25 -20
View File
@@ -41,19 +41,24 @@ async function getMongo() {
} }
async function store(event) { async function store(event) {
const r = await getRedis(); try {
await r.hSet("iot:latest", event.sensor_id, JSON.stringify({ const r = await getRedis();
value: event.value, unit: event.unit, location: event.location, timestamp: event.timestamp, await r.hSet("iot:latest", event.sensor_id, JSON.stringify({
})); value: event.value, unit: event.unit, location: event.location, timestamp: event.timestamp,
await r.hIncrBy("iot:counters", event.device_type, 1); }));
await r.zAdd("iot:recent", [{ score: 0, value: JSON.stringify(event) }]); await r.hIncrBy("iot:counters", event.device_type, 1);
await r.zRemRangeByRank("iot:recent", 0, -1001); await r.zAdd("iot:recent", [{ score: 0, value: JSON.stringify(event) }]);
const coll = await getMongo(); await r.zRemRangeByRank("iot:recent", 0, -1001);
await coll.insertOne({ } catch (e) { console.error("Redis error:", e.message); }
event_id: event.event_id, sensor_id: event.sensor_id,
device_type: event.device_type, location: event.location, try {
value: event.value, unit: event.unit, timestamp: new Date(event.timestamp), const coll = await getMongo();
}); await coll.insertOne({
event_id: event.event_id, sensor_id: event.sensor_id,
device_type: event.device_type, location: event.location,
value: event.value, unit: event.unit, timestamp: new Date(event.timestamp),
});
} catch (e) { console.error("Mongo error:", e.message); }
} }
async function start() { async function start() {
@@ -62,16 +67,16 @@ async function start() {
await ch.assertQueue(RMQ_QUEUE, { durable: true }); await ch.assertQueue(RMQ_QUEUE, { durable: true });
ch.prefetch(10); ch.prefetch(10);
await ch.consume(RMQ_QUEUE, async (msg) => { await ch.consume(RMQ_QUEUE, async (msg) => {
if (!msg) return;
try { try {
if (msg) { const event = JSON.parse(msg.content.toString());
const event = JSON.parse(msg.content.toString()); await store(event);
await store(event); consumed++;
consumed++; ch.ack(msg);
ch.ack(msg); if (consumed % 10 === 0) console.log(`Consumed ${consumed} events`);
if (consumed % 10 === 0) console.log(`Consumed ${consumed} events`);
}
} catch (e) { } catch (e) {
errors++; errors++;
ch.nack(msg, false, true);
console.error("Error:", e.message); console.error("Error:", e.message);
} }
}); });