Repeatedly try if a port on a host is open until duration seconds passed from: https://gist.github.com/betrcode/0248f0fda894013382d7#gistcomment-3161499 :param str host: host ip address or hostname :param int port: port number :param int/float duration: Optional. Total duration in
(host, port, duration=10, delay=2)
| 241 | |
| 242 | |
| 243 | async def wait_host_port(host, port, duration=10, delay=2): |
| 244 | """Repeatedly try if a port on a host is open until duration seconds passed |
| 245 | |
| 246 | from: https://gist.github.com/betrcode/0248f0fda894013382d7#gistcomment-3161499 |
| 247 | |
| 248 | :param str host: host ip address or hostname |
| 249 | :param int port: port number |
| 250 | :param int/float duration: Optional. Total duration in seconds to wait, by default 10 |
| 251 | :param int/float delay: Optional. Delay in seconds between each try, by default 2 |
| 252 | :return: awaitable bool |
| 253 | """ |
| 254 | tmax = time.time() + duration |
| 255 | while time.time() < tmax: |
| 256 | try: |
| 257 | _, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=5) |
| 258 | writer.close() |
| 259 | |
| 260 | # asyncio.StreamWriter.wait_closed is introduced in py 3.7 |
| 261 | # See https://docs.python.org/3/library/asyncio-stream.html#asyncio.StreamWriter.wait_closed |
| 262 | if hasattr(writer, 'wait_closed'): |
| 263 | await writer.wait_closed() |
| 264 | |
| 265 | return True |
| 266 | except Exception: |
| 267 | if delay: |
| 268 | await asyncio.sleep(delay) |
| 269 | return False |
| 270 | |
| 271 | |
| 272 | def get_free_port(): |
no test coverage detected
searching dependent graphs…