nocturn9x
/
giambio
mirror of https://github.com/nocturn9x/giambio.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Minor change to task_ipc test

This commit is contained in:
Nocturn9x 2022-02-27 13:40:26 +01:00
parent 3ec0864734
commit 800a173aa6
1 changed files with 6 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
tests/task_ipc.py
View File

@ -15,19 +15,19 @@ async def task(c: giambio.MemoryChannel, name: str):
await giambio.sleep(1)
async def main(channel: giambio.MemoryChannel):
print("[main] Spawning workers")
async with giambio.skip_after(5) as pool:
async def main(channel: giambio.MemoryChannel, delay: int):
print(f"[main] Spawning workers, exiting in {delay} seconds")
async with giambio.skip_after(delay) as pool:
await pool.spawn(task, channel, "one")
await pool.spawn(task, channel, "two")
await pool.spawn(task, channel, "three")
await channel.close()
print(f"[main] Operation complete, channel closed")
if await channel.pending():
print(f"[main] Channel has {len(channel.buffer)} leftover packets of data, clearing it")
print(f"[main] Channel has {len(channel.buffer)} leftover packet{'s' if len(channel.buffer) > 1 else ''} of data, clearing it")
while await channel.pending():
print(f"Cleared {await channel.read()!r}")
print(f"[main] Cleared {await channel.read()!r}")
channel = giambio.MemoryChannel()
giambio.run(main, channel, debugger=())
giambio.run(main, channel, 6, debugger=())