def Ctl.run

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

import Base
import bend-kube-runtime@0.2.0.0/src/json.bend as J
import bend-kube-runtime@0.2.0.0/src/config.bend as K
import bend-kube-runtime@0.2.0.0/src/queue.bend as Q

527# The controller loop: a message in, the state stepped, timers armed,
528# reconciles started, the next message awaited.
529def Ctl.run(~rec: U32 -> J.Json -> K.Config -> IO(Q.Action), fuel: Nat, ev: CEv, +st: CState,
530  +ctx: CCtx) -> IO(Unit):
531  match fuel ev:
532    case 0n e:
533      IO.pure(Unit, Unit{})
534    case 1n+f CMsg{None{}}:
535      IO.pure(Unit, Unit{})
536    case 1n+f CMsg{Some{m}}:
537      Ctl.run(~rec, f, CHandled{Ctl.on_msg(st, m, CCtx.kind(ctx))}, st, ctx)
538    case 1n+f CHandled{(+st1, ts)}:
539      IO.bind(Unit, Unit, Ctl.timers(ts, CCtx.chan(ctx)),
540        u => Ctl.run(~rec, f, CDisp{Ctl.dispatch(st1)}, st1, ctx))
541    case 1n+f CDisp{(+st2, jobs)}:
542      IO.bind(Unit, Unit, Ctl.spawn(~rec, jobs, CCtx.chan(ctx), CCtx.cfg(ctx), CCtx.idx(ctx)),
543        u => IO.bind(Result<&1, &1, SMsg, Unit>, Unit,
544          Chan.send(SMsg, CCtx.status(ctx), SPut{CCtx.idx(ctx), Ctl.snap(st2, U32.show(CCtx.idx(ctx)) ++ "/" ++ CCtx.kind(ctx))}),
545          u2 => IO.bind(Maybe<&1, Msg>, Unit, Chan.recv(Msg, CCtx.chan(ctx)),
546            m => Ctl.run(~rec, f, CMsg{m}, st2, ctx))))