(self, callback, *args, **kwargs)
| 65 | return error_sentinel |
| 66 | |
| 67 | async def _infinite_retry(self, callback, *args, **kwargs): |
| 68 | interval = kwargs.pop("_interval", 300) |
| 69 | context = kwargs.pop("_context", "") |
| 70 | # default overall timeout of 10 minutes (300 second interval * 2 iterations) |
| 71 | max_retries = kwargs.pop("_max_retries", 1) |
| 72 | if not context: |
| 73 | context = f"{callback.__name__}({args}, {kwargs})" |
| 74 | retries = 0 |
| 75 | while not self._shutdown_status: |
| 76 | try: |
| 77 | return await asyncio.wait_for(callback(*args, **kwargs), timeout=interval) |
| 78 | except (TimeoutError, asyncio.exceptions.TimeoutError): |
| 79 | self.log.debug(f"{self.name}: Timeout after {interval:,} seconds {context}, retrying...") |
| 80 | retries += 1 |
| 81 | if max_retries is not None and retries > max_retries: |
| 82 | raise TimeoutError(f"Timed out after {(max_retries + 1) * interval:,} seconds {context}") |
| 83 | |
| 84 | def engine_debug(self, *args, **kwargs): |
| 85 | if self._engine_debug: |
no test coverage detected