def Stream.loop

bend-kube-runtime@0.2.0.0 · src/stream.bend:87

import Base
import bend-kube-runtime@0.2.0.0/src/client.bend as Cl

87# Reads until the body ends, folding the body into st; bad holds the body
88# of a refusal. Answers the socket and the outcome.
89def Stream.loop(~S: Type, ~on: S -> List<&2, U32> -> IO(S), fuel: Nat, ev: SEv, st: S,
90  +bad: List<&2, U32>, +ms: U32) -> IO(Socket & Res(S)):
91  match fuel ev:
92    case 0n SGot{s, rp, r}:
93      IO.pure(Socket & Res(S), (s, Fail{Cl.KNet{0, "stream: too long"}}))
94    case 0n SChk{s, r, b, er, ok, h, em, fin, c}:
95      IO.pure(Socket & Res(S), (s, Fail{Cl.KNet{0, "stream: too long"}}))
96    case 1n+f SGot{s, rp, Done{+bs}}:
97      Stream.loop(~S, ~on, f, Stream.fed(s, bs, rp), st, bad, ms)
98    case 1n+f SGot{s, rp, Fail{(c, m)}}:
99      IO.pure(Socket & Res(S), (s, Fail{Cl.KNet{c, "stream: recv: " ++ m}}))
100    case 1n+f SChk{s, r, b, Some{m}, ok, h, em, fin, c}:
101      IO.pure(Socket & Res(S), (s, Fail{Cl.KNet{0, "stream: " ++ m}}))
102    case 1n+f SChk{s, +r, b, None{}, True{}, h, False{}, fin, +c}:
103      IO.bind(S, Socket & Res(S), on(st, b),
104        st2 => Stream.loop(~S, ~on, f, SChk{s, r, Nil{}, None{}, True{}, True{}, True{}, fin, c},
105          st2, bad, ms))
106    case 1n+f SChk{s, r, b, None{}, True{}, h, True{}, True{}, c}:
107      IO.pure(Socket & Res(S), (s, Done{st}))
108    case 1n+f SChk{s, +r, b, None{}, True{}, h, True{}, False{}, c}:
109      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>, Socket & Res(S),
110        Stream.recv(s, ms), p => Stream.loop(~S, ~on, f, SEv.got(p, r), st, bad, ms))
111    case 1n+f SChk{s, r, b, None{}, False{}, True{}, em, True{}, +c}:
112      IO.pure(Socket & Res(S), (s, Fail{Stream.refused(c, List.append(&2, U32, bad, b))}))
113    case 1n+f SChk{s, +r, b, None{}, False{}, True{}, em, False{}, c}:
114      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>, Socket & Res(S),
115        Stream.recv(s, ms),
116        p => Stream.loop(~S, ~on, f, SEv.got(p, r), st, List.append(&2, U32, bad, b), ms))
117    case 1n+f SChk{s, r, b, None{}, False{}, False{}, em, True{}, c}:
118      IO.pure(Socket & Res(S), (s, Fail{Cl.KNet{0, "stream: connection closed before a response"}}))
119    case 1n+f SChk{s, +r, b, None{}, False{}, False{}, em, False{}, c}:
120      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>, Socket & Res(S),
121        Stream.recv(s, ms), p => Stream.loop(~S, ~on, f, SEv.got(p, r), st, bad, ms))