MCPcopy Index your code
hub / github.com/pywebio/PyWebIO / wait_host_port

Function wait_host_port

pywebio/utils.py:243–269  ·  view source on GitHub ↗

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)

Source from the content-addressed store, hash-verified

241
242
243async 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
272def get_free_port():

Callers 3

run_testFunction · 0.90
corobasedFunction · 0.85

Calls 1

closeMethod · 0.45

Tested by

no test coverage detected

Used in the wild real call sites across dependent graphs

searching dependent graphs…