From 0bf1847abbaabc78637b41607dab22c400493ca3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= <tazet@narod.ru>
Date: Wed, 22 Jul 2026 08:00:16 +0400
Subject: [PATCH] feat: Node.js producer amqplib

---
 .trigger         |  0
 package.json     |  7 ++++++
 requirements.txt |  3 +++
 server.js        | 59 ++++++++++++++++++++++++++++++++++++++++++++++++
 4 files changed, 69 insertions(+)
 create mode 100644 .trigger
 create mode 100644 package.json
 create mode 100644 requirements.txt
 create mode 100644 server.js

diff --git a/.trigger b/.trigger
new file mode 100644
index 0000000..e69de29
diff --git a/package.json b/package.json
new file mode 100644
index 0000000..9ccbadb
--- /dev/null
+++ b/package.json
@@ -0,0 +1,7 @@
+{
+  "name": "iot-producer",
+  "version": "1.0.0",
+  "main": "server.js",
+  "scripts": { "start": "node server.js" },
+  "dependencies": { "amqplib": "^0.10.0", "express": "^4.21.0" }
+}
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..33b5482
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,3 @@
+Flask>=3.0
+gunicorn>=21.2
+kombu
diff --git a/server.js b/server.js
new file mode 100644
index 0000000..8a10613
--- /dev/null
+++ b/server.js
@@ -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(); });