Record client activity timestamp
(self, client_ip: str)
| 79 | self.lock = threading.Lock() |
| 80 | |
| 81 | def record_activity(self, client_ip: str): |
| 82 | """Record client activity timestamp""" |
| 83 | with self.lock: |
| 84 | prev_time = self.last_activity.get(client_ip) |
| 85 | current_time = time.time() |
| 86 | self.last_activity[client_ip] = current_time |
| 87 | if not prev_time: |
| 88 | logging.info(f"New client connected: {client_ip}") |
| 89 | else: |
| 90 | logging.debug(f"Client activity: {client_ip}") |
| 91 | |
| 92 | def cleanup_inactive(self, timeout: float) -> bool: |
| 93 | """Remove inactive clients""" |
no test coverage detected