def pump

bendonnect-rpc@0.1.0.0 · src/transport/client.bend:238

import Base
import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
import bendonnect-rpc@0.1.0.0/src/error.bend as E

238def pump(~on_msg: Bytes.Bytes -> IO(Unit), +collect: Bool, item: Item, acc: List<&1, Bytes.Bytes>) -> IO(Result<&1, &1, E.Error, Fin>):
239  match item:
240    case Msg{st, data, rest}:
241      do IO<Result<&1, &1, E.Error, Fin>>:
242        acc2 : List<&1, Bytes.Bytes> <- emit(~on_msg, collect, data, acc)
243        next : Item <- recv(st, rest)
244        pump(~on_msg, collect, next, acc2)
245    case Over{error, metadata}:
246      IO.pure(Result<&1, &1, E.Error, Fin>, end.done(error, metadata, acc))
247    case Broke{e}:
248      IO.pure(Result<&1, &1, E.Error, Fin>, Fail{e})