def Drv.run

bend-kube-runtime@0.2.0.0 · src/runtime.bend:241

import Base
import bend-kube-runtime@0.2.0.0/src/json.bend as J
import bend-kube-runtime@0.2.0.0/src/http.bend as H
import bend-kube-runtime@0.2.0.0/src/core.bend as C
import bend-kube-runtime@0.2.0.0/src/net.bend as N
import bend-kube-runtime@0.2.0.0/src/client.bend as Cl
import bend-kube-runtime@0.2.0.0/src/watcher.bend as W

241# The watcher's loop: list, then watch and stream, reconnecting and
242# relisting as the state machine says, until the fuel or the channel ends.
243def Drv.run(fuel: Nat, ev: DEv, +st: W.WState, +resp: H.Resp, +carry: List<&2, U32>,
244  +ctx: WCtx) -> IO(Unit):
245  match fuel ev:
246    case 0n e:
247      IO.pure(Unit, Unit{})
248    case 1n+f DStart{}:
249      match st:
250        case W.WState{W.WList{+c}, +fl, +first}:
251          IO.bind(Unit, Unit, Drv.note_list(first, WCtx.path(ctx)),
252            u => IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
253              Cl.Client.json(WCtx.cfg(ctx), WCtx.list_req(ctx, c)),
254              r => Drv.run(f, DListed{r}, W.WState{W.WList{c}, fl, first}, resp, carry, ctx)))
255        case W.WState{W.WWatch{+rv}, +fl, +first}:
256          IO.bind(Result<&1, &1, U32 & String, N.Tls>, Unit,
257            Cl.Conn.open(WCtx.cfg(ctx), WCtx.watch_req(ctx, rv), False{}),
258            r => Drv.run(f, DOpened{r}, W.WState{W.WWatch{rv}, fl, first}, resp, carry, ctx))
259    case 1n+f DListed{Done{+j}}:
260      Drv.run(f, DStepped{W.Watcher.step(st, W.IPage{C.Page.of(j)}), None{}, False{}}, st, resp,
261        carry, ctx)
262    case 1n+f DListed{Fail{+e}}:
263      Drv.run(f, DStepped{W.Watcher.step(st, W.IListErr{Cl.KErr.code(e)}), None{}, False{}}, st,
264        resp, carry, ctx)
265    case 1n+f DOpened{Done{N.Tls{s}}}:
266      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>, Unit,
267        N.Tls.recv(s, 65536, 350000),
268        p => Drv.run(f, DEv.recv(p), st, H.Http.resp.init(), Nil{}, ctx))
269    case 1n+f DOpened{Fail{e}}:
270      Drv.run(f, DStepped{W.Watcher.step(st, W.IWatchErr{0}), None{}, False{}}, st, resp, carry,
271        ctx)
272    case 1n+f DRecv{s, Done{+bs}}:
273      Drv.run(f, DDig{s, Drv.digest(st, H.Http.feed(bs, resp), carry,
274        List.is_empty(&2, U32, bs))}, st, resp, carry, ctx)
275    case 1n+f DDig{s, Dig{step, r2, c2, more}}:
276      Drv.run(f, DStepped{step, Some{s}, more}, st, r2, c2, ctx)
277    case 1n+f DRecv{s, Fail{e}}:
278      Drv.run(f, DStepped{W.Watcher.step(st, W.IWatchErr{0}), Some{s}, False{}}, st, resp,
279        carry, ctx)
280    case 1n+f DStepped{W.Step{st2, outs, wait}, s, more}:
281      IO.bind(Unit, Unit, Drv.send_all(WCtx.subs(ctx), outs),
282        u => Drv.run(f, DNext{s, wait, more}, st2, resp, carry, ctx))
283    case 1n+f DNext{Some{s}, False{}, True{}}:
284      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>, Unit,
285        N.Tls.recv(s, 65536, 350000), p => Drv.run(f, DEv.recv(p), st, resp, carry, ctx))
286    case 1n+f DNext{s, wait, more}:
287      IO.bind(Unit, Unit, Drv.close(s),
288        u => IO.bind(Unit, Unit, Drv.pause(wait, WState.fails(st), WCtx.path(ctx)),
289          u2 => Drv.run(f, DStart{}, st, H.Http.resp.init(), Nil{}, ctx)))