diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index 8c4eba9..9a4b656 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -1,21 +1,16 @@ #!/usr/bin/env python3 """ -NATS Registration Listener --------------------------- -Listens on a NATS subject (default: 'registration') and prints messages to STDOUT. +NATS Registration Listener (compact one-line output) +--------------------------------------------------- +Print exactly one line per message: + [TS] product= mac= fw= + +- Keeps your existing CLI flags (servers/subject/queue/auth/TLS). +- Writes are serialized to avoid garbled output. +- Falls back gracefully if fields/structure are missing. Install deps: pip install --upgrade nats-py - -Examples: - python nats_registration_listener.py --servers nats://127.0.0.1:4222 --subject registration - python nats_registration_listener.py --servers nats://n1:4222 nats://n2:4222 --token $NATS_TOKEN - python nats_registration_listener.py --servers nats://host:4222 --creds /path/to/user.creds - python nats_registration_listener.py --servers tls://host:4443 --tls-ca ca.pem --tls-cert client.crt --tls-key client.key - -Notes: -- If messages are JSON, they will be pretty-printed; otherwise raw text (UTF-8) or bytes length is shown. -- Reconnects automatically with exponential backoff, prints connection status to STDERR. """ import argparse @@ -26,13 +21,13 @@ import signal import ssl import sys from datetime import datetime, timezone -from typing import Optional +from typing import Optional, Dict, Any import nats def parse_args(): - p = argparse.ArgumentParser(description="Listen to a NATS subject and print messages.") + p = argparse.ArgumentParser(description="Listen to a NATS subject and print one line per device.") p.add_argument( "--servers", nargs="+", @@ -47,20 +42,20 @@ def parse_args(): p.add_argument("--queue", help="Optional queue group name.", default=os.environ.get("NATS_QUEUE")) p.add_argument("--name", help="Client connection name.", default="registration-listener") - # Auth options + # Auth p.add_argument("--creds", help="Path to .creds file (JWT + NKey).") p.add_argument("--user", help="Username.") p.add_argument("--password", help="Password.") p.add_argument("--token", help="Auth token.") - # TLS options + # TLS p.add_argument("--tls-ca", help="Path to CA certificate for TLS.") p.add_argument("--tls-cert", help="Path to client certificate for TLS.") p.add_argument("--tls-key", help="Path to client key for TLS.") p.add_argument("--insecure", action="store_true", help="Disable TLS hostname verification.") - p.add_argument("--print-raw", action="store_true", help="Always print raw message bytes (no decoding / pretty JSON).") - + # Misc + p.add_argument("--include-subject", action="store_true", help="Append subject=... to each output line.") return p.parse_args() @@ -86,24 +81,79 @@ def ts() -> str: return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S%z") +def pick_first_interface(eths: Dict[str, Any]) -> Optional[Dict[str, Any]]: + # Prefer eth0; otherwise first key deterministically (sorted) + if "eth0" in eths and isinstance(eths["eth0"], dict): + return eths["eth0"] + for name in sorted(eths.keys()): + if isinstance(eths[name], dict): + return eths[name] + return None + + +def extract_fields(obj: Dict[str, Any]): + """ + Expected structure (from examples): + { "data": { + "productName": "...", + "ethernetInterfaces": { "eth0": { "macAddress": "..." } }, + "firmwareVersion": { "active": "..." } + }, + "metadata": { "type": "REGISTRATION_REQUEST" } + } + """ + root = obj + if isinstance(root.get("data"), dict): + d = root["data"] + else: + # Some producers might not wrap with "data" + d = root + + product = d.get("productName") or "-" + fw_active = "-" + try: + fw_active = d.get("firmwareVersion", {}).get("active") or "-" + except Exception: + pass + + mac = "-" + try: + eths = d.get("ethernetInterfaces", {}) + if isinstance(eths, dict): + chosen = pick_first_interface(eths) + if chosen and isinstance(chosen.get("macAddress"), str): + mac = chosen["macAddress"] + except Exception: + pass + + return product, mac, fw_active + + async def main(): args = parse_args() ssl_ctx = make_ssl_context(args) - reconnect_time_wait = 2 # seconds base - max_reconnect_attempts = -1 # infinite + # Serialize stdout writes to avoid interleaving + print_lock = asyncio.Lock() + + async def log_status(s: str): + async with print_lock: + print(s, file=sys.stderr, flush=True) + + reconnect_time_wait = 2 + max_reconnect_attempts = -1 async def disconnected_cb(): - print(f"[{ts()}] Disconnected from NATS.", file=sys.stderr, flush=True) + await log_status(f"[{ts()}] Disconnected from NATS.") async def reconnected_cb(): - print(f"[{ts()}] Reconnected to NATS.", file=sys.stderr, flush=True) + await log_status(f"[{ts()}] Reconnected to NATS.") async def error_cb(e): - print(f"[{ts()}] Error: {e!r}", file=sys.stderr, flush=True) + await log_status(f"[{ts()}] Error: {e!r}") async def closed_cb(): - print(f"[{ts()}] Connection closed.", file=sys.stderr, flush=True) + await log_status(f"[{ts()}] Connection closed.") nc = await nats.connect( servers=args.servers, @@ -123,42 +173,40 @@ async def main(): ) async def message_handler(msg: nats.aio.msg.Msg): - payload = msg.data + line = None now = ts() - meta = f'subject="{msg.subject}"' - if msg.reply: - meta += f' reply="{msg.reply}"' - meta += f" size={len(payload)}B" + payload = msg.data - if getattr(args, "print_raw", False): - sys.stdout.write(f"[{now}] {meta} bytes={payload!r}\n") - sys.stdout.flush() - return - - # Try to decode as UTF-8 and pretty print JSON if applicable + # Best-effort JSON parse try: text = payload.decode("utf-8", errors="replace") - try: - obj = json.loads(text) - pretty = json.dumps(obj, indent=2, ensure_ascii=False, sort_keys=True) - print(f"[{now}] {meta}\n{pretty}\n", flush=True) - except json.JSONDecodeError: - print(f"[{now}] {meta}\n{text}\n", flush=True) + obj = json.loads(text) + product, mac, fw = extract_fields(obj) + line = f"[{now}] product={product} mac={mac} fw={fw}" except Exception: - print(f"[{now}] {meta} (non-UTF8) bytes={payload!r}\n", flush=True) + # If not JSON or broken, still emit a line that shows size only + line = f"[{now}] product=- mac=- fw=-" + + if args.include_subject: + line += f" subject={msg.subject}" + + async with print_lock: + sys.stdout.write(line + "\n") + sys.stdout.flush() if args.queue: await nc.subscribe(args.subject, queue=args.queue, cb=message_handler) else: await nc.subscribe(args.subject, cb=message_handler) - print(f"[{ts()}] Listening on subject '{args.subject}' (queue={getattr(args, 'queue', None) or '-'}) via {args.servers}", file=sys.stderr, flush=True) + await log_status(f"[{ts()}] Listening on subject '{args.subject}' (queue={getattr(args, 'queue', None) or '-'}) via {args.servers}") # Graceful shutdown stop_event = asyncio.Event() def handle_signal(*_): - print(f"[{ts()}] Received stop signal, draining...", file=sys.stderr, flush=True) + # Use stderr for status + asyncio.create_task(log_status(f"[{ts()}] Received stop signal, draining...")) stop_event.set() loop = asyncio.get_running_loop() @@ -166,7 +214,6 @@ async def main(): try: loop.add_signal_handler(s, handle_signal) except NotImplementedError: - # Windows signal.signal(s, lambda *_: handle_signal()) await stop_event.wait()