(self, path: Path, socketio: SocketIO, shutdown_event: threading.Event)
| 409 | |
| 410 | class LogSession: |
| 411 | def __init__(self, path: Path, socketio: SocketIO, shutdown_event: threading.Event): |
| 412 | self.path = path |
| 413 | self.socketio = socketio |
| 414 | self.shutdown_event = shutdown_event |
| 415 | self.room = _room_for_path(path) |
| 416 | self.parser = LogParser() |
| 417 | self.lock = threading.Lock() |
| 418 | self.subscribers: Set[str] = set() |
| 419 | self.missing = False |
| 420 | self._watching = False |
| 421 | self._thread: Optional[threading.Thread] = None |
| 422 | self._last_position = 0 |
| 423 | self._last_size = 0 |
| 424 | self._last_inode: Optional[int] = None |
| 425 | self._loaded = False |
| 426 | |
| 427 | def _reset_parser(self) -> None: |
| 428 | with self.lock: |
nothing calls this directly
no test coverage detected