one liners 0600
This commit is contained in:
@@ -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=<productName> mac=<mac> fw=<active>
|
||||
|
||||
- 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()
|
||||
|
||||
Reference in New Issue
Block a user