def select.relay

bend-kit-concurrency@0.2.0.0 · concurrency.bend:249

import Base

249# Forwards each value of src to out as (i, value), then reports on done.
250# It stops when src closes, or when out is closed under it.
251def select.relay(
252  -A: Type, fuel: Nat, +i: U32, +src: Chan(A), +out: Chan(U32 & A),
253  +done: Chan(Unit), r: Relay<A>
254) -> IO(Result<&1, &1, Unit, Unit>):
255  match fuel:
256    case 0n:
257      Chan.send(Unit, done, Unit{})
258    case 1n+p:
259      match r:
260        case RWait{}:
261          do IO<Result<&1, &1, Unit, Unit>>:
262            got : Maybe<&1, A> <- Chan.recv(A, src)
263            select.relay(A, p, i, src, out, done, RGot{got})
264        case RGot{got}:
265          match got:
266            case None{}:
267              Chan.send(Unit, done, Unit{})
268            case Some{x}:
269              do IO<Result<&1, &1, Unit, Unit>>:
270                ok : Result<&1, &1, U32 & A, Unit> <- Chan.send(U32 & A, out, (i, x))
271                select.relay(A, p, i, src, out, done, RSent{ok})
272        case RSent{ok}:
273          match ok:
274            case Done{u}:
275              select.relay(A, p, i, src, out, done, RWait{})
276            case Fail{held}:
277              Chan.send(Unit, done, Unit{})