(self)
| 220 | ] |
| 221 | |
| 222 | def _capture_loop(self) -> None: |
| 223 | try: |
| 224 | self._proc = subprocess.Popen( |
| 225 | self._build_cmd(), |
| 226 | stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, |
| 227 | text=True, bufsize=1, |
| 228 | ) |
| 229 | except Exception as exc: |
| 230 | logger.error("Failed to spawn tshark for IRC DPI: %s", exc) |
| 231 | self._running = False |
| 232 | return |
| 233 | |
| 234 | try: |
| 235 | assert self._proc.stdout is not None |
| 236 | for line in self._proc.stdout: |
| 237 | if not self._running: |
| 238 | break |
| 239 | self._handle_tshark_line(line.rstrip('\n')) |
| 240 | finally: |
| 241 | self.stop() |
| 242 | |
| 243 | def _handle_tshark_line(self, line: str) -> None: |
| 244 | parts = line.split('\t') |
nothing calls this directly
no test coverage detected