()
| 67 | await nconn.execute("notify foo, '2'") |
| 68 | |
| 69 | async def receiver(): |
| 70 | await aconn.set_autocommit(True) |
| 71 | cur = aconn.cursor() |
| 72 | await cur.execute("listen foo") |
| 73 | gen = aconn.notifies() |
| 74 | async for n in gen: |
| 75 | ns.append((n, time())) |
| 76 | if len(ns) >= 2: |
| 77 | await gen.aclose() |
| 78 | |
| 79 | ns: list[tuple[Notify, float]] = [] |
| 80 | t0 = time() |
nothing calls this directly
no test coverage detected