def talk.read.loop

bend-kit-http@0.38.2.0 · http.bend:6099

import Base
import bend-kit-http@0.38.2.0/admission.bend as Admission
import bend-kit-wire@0.4.7.0/wire.bend as Wire
import bend-kit-time@0.3.0.0/time.bend as Time

6099def talk.read.loop(+inbox: Chan(Admission.Command), +held: Bool, +settings: Limits,
6100  step: ReadStep) -> IO((Socket & Turn) & Phase):
6101  match step:
6102    case ReadChecking{s, sv, phase}:
6103      do IO<(Socket & Turn) & Phase>:
6104        running : Bool <- talk.read.running(held, inbox)
6105        talk.read.loop(inbox, held, settings, ReadAllowed{running, s, sv, phase})
6106    case ReadAllowed{running, s, sv, +phase}:
6107      match running:
6108        case False{}:
6109          IO.pure((Socket & Turn) & Phase, ((s, TurnClose{}), phase))
6110        case True{}:
6111          Phase{kind, until} = phase
6112          do IO<(Socket & Turn) & Phase>:
6113            now : Time.Duration <- Time.mono()
6114            +ms : U32 = phase.remaining(until, now)
6115            talk.read.loop(inbox, held, settings,
6116              ReadBudget{Bool.pick(U32, held, ms, U32.min(ms, 50)), s, sv, phase})
6117    case ReadBudget{ms, s, sv, +phase}:
6118      match ms:
6119        case 0:
6120          IO.pure((Socket & Turn) & Phase, ((s, phase.expired(phase)), phase))
6121        case _:
6122          do IO<(Socket & Turn) & Phase>:
6123            got : Socket & Result<&1, &1, U32 & String, U32 & Array<U32>> <- Wire.recv.words(s, 65536, ms)
6124            now : Time.Duration <- Time.mono()
6125            talk.read.loop(inbox, held, settings, ReadReceived{sv, phase, now, got})
6126    case ReadReceived{sv, +phase, +now, got}:
6127      Phase{kind, until} = phase
6128      (s, result) = got
6129      match result:
6130        case Fail{(+code, message)}:
6131          talk.read.loop(inbox, held, settings,
6132            ReadFailed{err.late(code) && Cmp.is_lt(Time.Duration.cmp(now, until)), s, sv, phase, now, (code, message)})
6133        case Done{packed}:
6134          IO.pure((Socket & Turn) & Phase,
6135            talk.received.late(Bool.not(Cmp.is_lt(Time.Duration.cmp(now, until))), settings, phase, now, sv, (s, Done{packed})))
6136    case ReadFailed{retry, s, sv, +phase, +now, error}:
6137      match retry:
6138        case True{}:
6139          talk.read.loop(inbox, held, settings, ReadChecking{s, sv, phase})
6140        case False{}:
6141          Phase{kind, until} = phase
6142          IO.pure((Socket & Turn) & Phase,
6143            talk.received.late(Bool.not(Cmp.is_lt(Time.Duration.cmp(now, until))), settings, phase, now, sv, (s, Fail{error})))