93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
"""Consumer — RabbitMQ HTTP API → Redis + MongoDB (no AMQP libs)"""
|
|
import json, os, base64, threading, time
|
|
from urllib.request import Request, urlopen
|
|
from flask import Flask, jsonify
|
|
import redis, pymongo
|
|
|
|
app = Flask(__name__)
|
|
|
|
RMQ_HOST = os.environ["RMQ_HOST"]
|
|
RMQ_USER = os.environ["RMQ_USER"]
|
|
RMQ_PASS = os.environ["RMQ_PASS"]
|
|
RMQ_QUEUE = os.environ["RMQ_QUEUE"]
|
|
RMQ_API = f"http://{RMQ_HOST}:15672/api"
|
|
AUTH = base64.b64encode(f"{RMQ_USER}:{RMQ_PASS}".encode()).decode()
|
|
|
|
REDIS_HOST = os.environ["REDIS_HOST"]
|
|
REDIS_PORT = int(os.environ.get("REDIS_PORT", 6379))
|
|
REDIS_PASS = os.environ.get("REDIS_PASS", "")
|
|
|
|
MONGO_URI = os.environ["MONGO_URI"]
|
|
MONGO_DB = os.environ.get("MONGO_DB", "iot")
|
|
|
|
consumed = 0; errors = 0
|
|
r = None; mongo = None
|
|
|
|
def get_redis():
|
|
global r
|
|
if r is None:
|
|
r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASS or None, decode_responses=True)
|
|
print("[CONSUMER] Redis connected")
|
|
return r
|
|
|
|
def get_mongo():
|
|
global mongo
|
|
if mongo is None:
|
|
mongo = pymongo.MongoClient(MONGO_URI)[MONGO_DB]["iot_events"]
|
|
mongo.create_index("timestamp", expireAfterSeconds=604800)
|
|
print("[CONSUMER] MongoDB connected")
|
|
return mongo
|
|
|
|
def rmq_get(url):
|
|
req = Request(url, headers={"Authorization": f"Basic {AUTH}"})
|
|
return json.loads(urlopen(req, timeout=5).read())
|
|
|
|
def store(event):
|
|
rd = get_redis()
|
|
rd.hset("iot:latest", event["sensor_id"], json.dumps({
|
|
"value": event["value"], "unit": event["unit"], "location": event["location"], "timestamp": event["timestamp"],
|
|
}))
|
|
rd.hincrby("iot:counters", event["device_type"], 1)
|
|
rd.zadd("iot:recent", {json.dumps(event): 0})
|
|
rd.zremrangebyrank("iot:recent", 0, -1001)
|
|
get_mongo().insert_one({
|
|
"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": event["timestamp"],
|
|
})
|
|
|
|
def poll_loop():
|
|
global consumed, errors
|
|
# Wait for RabbitMQ to be ready
|
|
time.sleep(10)
|
|
while True:
|
|
try:
|
|
# Get messages from queue via HTTP API
|
|
msgs = rmq_get(f"{RMQ_API}/queues/%2F/{RMQ_QUEUE}/get")
|
|
if msgs:
|
|
for m in msgs:
|
|
if m.get("payload"):
|
|
event = json.loads(base64.b64decode(m["payload"]).decode())
|
|
store(event)
|
|
consumed += 1
|
|
# Ack via API
|
|
rmq_get(f"{RMQ_API}/queues/%2F/{RMQ_QUEUE}/get?requeue=false&count=1")
|
|
if consumed > 0 and consumed % 10 == 0:
|
|
print(f"[CONSUMER] Processed {consumed}")
|
|
except Exception as e:
|
|
errors += 1
|
|
print(f"[CONSUMER] Error: {e}")
|
|
time.sleep(2)
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return jsonify({"service":"iot-consumer","status":"running","queue":RMQ_QUEUE,"consumed":consumed,"errors":errors})
|
|
|
|
@app.route("/health")
|
|
def health():
|
|
return jsonify({"status":"ok","consumed":consumed})
|
|
|
|
if __name__ == "__main__":
|
|
threading.Thread(target=poll_loop, daemon=True).start()
|
|
app.run(host="0.0.0.0", port=int(os.environ.get("PORT",5000)))
|