diff --git a/collect.py b/collect.py index 6f5bbe6..cdd8ea3 100644 --- a/collect.py +++ b/collect.py @@ -95,7 +95,7 @@ def run(args: argparse.Namespace) -> None: logs_dir.mkdir(exist_ok=True) out = auto_out_path(logs_dir, vin) - push_clients, push_lock = network.start_push_server(args.host, args.port) + push_clients, push_lock, push_last = network.start_push_server(args.host, args.port) current_day = datetime.now(timezone.utc).date() store = load_store(out, vin, args.interval) @@ -121,7 +121,7 @@ def run(args: argparse.Namespace) -> None: snapshot = collect_snapshot(vehicle, domains) store["records"].append(snapshot) save_store(out, store, args.max_records) - network.broadcast(snapshot, push_clients, push_lock) + network.broadcast(snapshot, push_clients, push_lock, push_last) log.info("Record #%d saved at %s", len(store["records"]), snapshot["ts"]) except KeyboardInterrupt: raise diff --git a/network.py b/network.py index c0309c9..8ccee12 100644 --- a/network.py +++ b/network.py @@ -10,20 +10,30 @@ def _accept_loop( server_sock: socket.socket, clients: list, lock: threading.Lock, + last_snapshot: dict, ) -> None: while True: try: conn, addr = server_sock.accept() log.info("Push client connected from %s", addr) with lock: + if last_snapshot["data"] is not None: + try: + line = (json.dumps(last_snapshot["data"], default=str) + "\n").encode() + conn.sendall(line) + except OSError: + log.debug("Failed to send initial snapshot to %s", addr) + conn.close() + continue clients.append(conn) except OSError: break -def broadcast(snapshot: dict, clients: list, lock: threading.Lock) -> None: +def broadcast(snapshot: dict, clients: list, lock: threading.Lock, last_snapshot: dict) -> None: line = (json.dumps(snapshot, default=str) + "\n").encode() with lock: + last_snapshot["data"] = snapshot dead = [] for conn in clients: try: @@ -43,7 +53,12 @@ def start_push_server(host: str, port: int) -> tuple: server_sock.listen() clients: list = [] lock = threading.Lock() - t = threading.Thread(target=_accept_loop, args=(server_sock, clients, lock), daemon=True) + last_snapshot: dict = {"data": None} + t = threading.Thread( + target=_accept_loop, + args=(server_sock, clients, lock, last_snapshot), + daemon=True, + ) t.start() log.info("Push server listening on %s:%d", host, port) - return clients, lock \ No newline at end of file + return clients, lock, last_snapshot \ No newline at end of file