| 320 | |
| 321 | @contextlib.contextmanager |
| 322 | def capture_signals(self) -> Generator[None, None, None]: |
| 323 | # Signals can only be listened to from the main thread. |
| 324 | if threading.current_thread() is not threading.main_thread(): |
| 325 | yield |
| 326 | return |
| 327 | # always use signal.signal, even if loop.add_signal_handler is available |
| 328 | # this allows to restore previous signal handlers later on |
| 329 | original_handlers = {sig: signal.signal(sig, self.handle_exit) for sig in HANDLED_SIGNALS} |
| 330 | try: |
| 331 | yield |
| 332 | finally: |
| 333 | for sig, handler in original_handlers.items(): |
| 334 | signal.signal(sig, handler) |
| 335 | # If we did gracefully shut down due to a signal, try to |
| 336 | # trigger the expected behaviour now; multiple signals would be |
| 337 | # done LIFO, see https://stackoverflow.com/questions/48434964 |
| 338 | for captured_signal in reversed(self._captured_signals): |
| 339 | signal.raise_signal(captured_signal) |
| 340 | |
| 341 | def handle_exit(self, sig: int, frame: FrameType | None) -> None: |
| 342 | self._captured_signals.append(sig) |