Forechan
go style asyncio channels.
Fully Typed with Generics, mypy ready.
Inspired by core.async from clojure.
Examples
Send & Recv
ch = chan(int) # Chan[int]
await (ch << 2) # or use `await ch.send(x)`
two = await ([] << ch) # or use `await ch.recv()`
assert two == 2
# `ch.close()` is idempotent
await ch.close()Basic
if ch:
# if `Chan[T]` is open
len(ch) # How many items are in `ch`
async with ch:
# close `ch` after exiting lexical-scope
async for item in ch:
# use `ch` as AsyncIteratorSelect
async for ch, item in select(ch1, ch2, ch3, ...):
if ch == ch1:
...
elif ch == ch2:
...
elif ch == ch3:
...Wait Group
wg = wait_group()
for _ in range(5):
async def cont() -> None:
with wg:
# do some work
create_task(cont())
# will wait for all work to be completed
await wg.wait()Synchronous
head = ch.try_peek() # can throw `ChanEmpty`
(ch < 2) # or use `ch.try_send(2)` , can throw `ChanFull`
two = ([] < ch) # or use `ch.try_recv()` , can throw `ChanEmpty`
assert two == 2Go -> Python
The following are roughly equivalent
func fn() {
// do things here
}
go fn()When GOMAXPROCS=1
async def fn() -> None:
# do things here
create_task(fn())Common Concurrency Patterns
Consumer
async def consumer() -> None:
async for item in ch:
# do something with `item`, until `ch` is closedProducer
def producer() -> Chan[int]:
ch = chan(int)
async def cont() -> None:
# auto close `ch` when done
async with ch:
while ...:
# send result `item` to downstream `ch`
await (ch << item)
create_task(cont())
return ch
