Egutierrez
49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Publisher NATS como proceso del sistema operativo independiente.
|
|
|
|
Se conecta al broker y publica una rafaga de mensajes de telemetria,
|
|
alternando entre los subjects `telemetria.cpu` y `telemetria.mem`.
|
|
No sabe ni le importa cuantos subscribers hay escuchando: solo conoce el
|
|
subject. Emite cada publicacion como linea JSON en stdout.
|
|
"""
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import random
|
|
import time
|
|
|
|
import nats
|
|
|
|
NATS_URL = "nats://127.0.0.1:4222"
|
|
|
|
|
|
def emit(event: dict) -> None:
|
|
print(json.dumps(event), flush=True)
|
|
|
|
|
|
async def main(count: int, interval: float) -> None:
|
|
pid = os.getpid()
|
|
nc = await nats.connect(NATS_URL, name="publisher")
|
|
emit({"event": "ready", "pid": pid, "name": "publisher"})
|
|
|
|
for i in range(count):
|
|
subject = "telemetria.cpu" if i % 2 == 0 else "telemetria.mem"
|
|
payload = json.dumps({"i": i, "valor": round(random.uniform(0, 100), 1)})
|
|
await nc.publish(subject, payload.encode())
|
|
emit({"event": "published", "pid": pid, "subject": subject, "i": i})
|
|
await asyncio.sleep(interval)
|
|
|
|
await nc.flush()
|
|
emit({"event": "done", "pid": pid, "name": "publisher", "published": count})
|
|
await nc.drain()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Publisher NATS de demostracion")
|
|
parser.add_argument("--count", type=int, default=8, help="Numero de mensajes a publicar")
|
|
parser.add_argument("--interval", type=float, default=0.15,
|
|
help="Segundos entre publicaciones")
|
|
args = parser.parse_args()
|
|
asyncio.run(main(args.count, args.interval))
|