(self, client_ip: str, server_ip: str,
server_port: int, ts: float,
msg: IRCMessage)
| 335 | return out |
| 336 | |
| 337 | def _update_session(self, client_ip: str, server_ip: str, |
| 338 | server_port: int, ts: float, |
| 339 | msg: IRCMessage) -> IRCSession: |
| 340 | skey = (client_ip, server_ip, server_port) |
| 341 | with self._lock: |
| 342 | sess = self._sessions.get(skey) |
| 343 | if sess is None: |
| 344 | if len(self._sessions) >= self.MAX_SESSIONS: |
| 345 | # Drop oldest by last_seen |
| 346 | oldest = min(self._sessions.values(), key=lambda s: s.last_seen) |
| 347 | self._sessions.pop( |
| 348 | (oldest.client_ip, oldest.server_ip, oldest.server_port), |
| 349 | None, |
| 350 | ) |
| 351 | sess = IRCSession( |
| 352 | client_ip=client_ip, server_ip=server_ip, |
| 353 | server_port=server_port, |
| 354 | first_seen=ts, last_seen=ts, |
| 355 | ) |
| 356 | self._sessions[skey] = sess |
| 357 | sess.last_seen = ts |
| 358 | sess.cmd_counts[msg.command] = sess.cmd_counts.get(msg.command, 0) + 1 |
| 359 | if msg.command in _INTERESTING_CMDS: |
| 360 | sess.messages.append(msg) |
| 361 | self._apply_message(sess, msg) |
| 362 | return sess |
| 363 | |
| 364 | @staticmethod |
| 365 | def _apply_message(sess: IRCSession, msg: IRCMessage) -> None: |
no test coverage detected