register the system shutdown trigger event
()
| 166 | |
| 167 | |
| 168 | async def get_shutdown_trigger() -> Callable[[], Coroutine[Any, Any, Any]]: |
| 169 | """register the system shutdown trigger event""" |
| 170 | signal_event = asyncio.Event() |
| 171 | loop = asyncio.get_event_loop() |
| 172 | |
| 173 | def _signal_handler(*_: Any) -> None: # noqa: N803 |
| 174 | print('receive signal event ...') |
| 175 | signal_event.set() |
| 176 | |
| 177 | for signal_name in ["SIGINT", "SIGTERM", "SIGBREAK"]: |
| 178 | if hasattr(signal, signal_name): |
| 179 | try: |
| 180 | loop.add_signal_handler( |
| 181 | getattr(signal, signal_name), _signal_handler) |
| 182 | except NotImplementedError: |
| 183 | # Add signal handler may not be implemented on Windows |
| 184 | signal.signal(getattr(signal, signal_name), _signal_handler) |
| 185 | |
| 186 | return signal_event.wait |
| 187 | |
| 188 | |
| 189 | async def shutdown(trigger: Callable[[], Coroutine[Any, Any, Any]]) -> None: |