def talk.read.loop

bend-kit-http@0.38.5.0 · http.bend:6091

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

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