feat: Node.js producer amqplib

2026-07-22 08:00:16 +04:00
commit 0bf1847abb
4 changed files with 69 additions and 0 deletions
@@ -0,0 +1,59 @@
+const amqp = require("amqplib");
+const express = require("express");
+
+const HOST = process.env.RABBIT_HOST || "localhost";
+const PORT = parseInt(process.env.RABBIT_PORT || "5672");
+const USER = process.env.RABBIT_USER || "guest";
+const PASS = process.env.RABBIT_PASS || "guest";
+const QUEUE = process.env.RABBIT_QUEUE || "iot-events";
+const INTERVAL = parseInt(process.env.PRODUCE_INTERVAL || "3") * 1000;
+
+const SENSORS = [
+  { id: "temp_living", type: "temperature", location: "living_room", unit: "°C", min: 18, max: 28 },
+  { id: "temp_bedroom", type: "temperature", location: "bedroom", unit: "°C", min: 16, max: 26 },
+  { id: "temp_office", type: "temperature", location: "office", unit: "°C", min: 19, max: 27 },
+  { id: "humidity_living", type: "humidity", location: "living_room", unit: "%", min: 30, max: 70 },
+  { id: "humidity_bed", type: "humidity", location: "bedroom", unit: "%", min: 35, max: 65 },
+  { id: "power_kitchen", type: "power_meter", location: "kitchen", unit: "kW", min: 0.1, max: 3.5 },
+  { id: "power_living", type: "power_meter", location: "living_room", unit: "kW", min: 0.2, max: 2 },
+  { id: "power_office", type: "power_meter", location: "office", unit: "kW", min: 0.3, max: 4 },
+];
+
+let channel = null;
+let eventsSent = 0;
+let errors = 0;
+
+async function connect() {
+  const conn = await amqp.connect(`amqp://${USER}:${PASS}@${HOST}:${PORT}`);
+  channel = await conn.createChannel();
+  await channel.assertQueue(QUEUE, { durable: true });
+  console.log(`Producer connected to RabbitMQ, queue: ${QUEUE}`);
+}
+
+async function produce() {
+  while (true) {
+    try {
+      if (!channel) await connect();
+      const s = SENSORS[Math.floor(Math.random() * SENSORS.length)];
+      const event = {
+        event_id: crypto.randomUUID(),
+        sensor_id: s.id, device_type: s.type, location: s.location,
+        value: +(s.min + Math.random() * (s.max - s.min)).toFixed(2),
+        unit: s.unit,
+        timestamp: new Date().toISOString(),
+      };
+      channel.sendToQueue(QUEUE, Buffer.from(JSON.stringify(event)), { persistent: true });
+      eventsSent++;
+      if (eventsSent % 10 === 0) console.log(`Sent ${eventsSent} events`);
+    } catch (e) {
+      errors++;
+      console.error("Error:", e.message);
+    }
+    await new Promise(r => setTimeout(r, INTERVAL));
+  }
+}
+
+const app = express();
+app.get("/", (req, res) => res.json({ service: "iot-producer", status: "running", queue: QUEUE, events_sent: eventsSent, errors }));
+app.get("/health", (req, res) => res.json({ status: "ok" }));
+app.listen(process.env.PORT || 3000, () => { console.log("Producer started"); produce(); });