def Pool.run

bend-kube-runtime@0.2.0.0 · src/client.bend:448

import Base
import bend-kube-runtime@0.2.0.0/src/http.bend as H
import bend-kube-runtime@0.2.0.0/src/config.bend as K
import bend-kube-runtime@0.2.0.0/src/net.bend as N

448# One request through the pool (fuel bounds the retry).
449def Pool.run(fuel: Nat, ev: PEv, +cfg: K.Config, +pool: Chan(Socket), +req: H.Req) ->
450  IO(Result<&2, &2, KErr, Reply>):
451  match fuel ev:
452    case 0n PTook{Ready{Some{s}}}:
453      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u => Pool.fail("pool: out of steps"))
454    case 0n PSent{u, Done{N.Tls{s}}}:
455      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u2 => Pool.fail("pool: out of steps"))
456    case 0n PRead{u, (s, r)}:
457      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u2 => Pool.fail("pool: out of steps"))
458    case 0n PFin{s, r, k, t}:
459      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s), u2 => Pool.fail("pool: out of steps"))
460    case 0n e:
461      Pool.fail("pool: out of steps")
462    case 1n+f PTook{Ready{Some{s}}}:
463      IO.bind(Result<&1, &1, U32 & String, N.Tls>, Result<&2, &2, KErr, Reply>,
464        Conn.send_on(cfg, s, req), r => Pool.run(f, PSent{True{}, r}, cfg, pool, req))
465    case 1n+f PTook{m}:
466      IO.bind(Result<&1, &1, U32 & String, N.Tls>, Result<&2, &2, KErr, Reply>,
467        Conn.open(cfg, req, True{}), r => Pool.run(f, PSent{False{}, r}, cfg, pool, req))
468    case 1n+f PSent{+reused, Done{N.Tls{s}}}:
469      IO.bind(Socket & H.Resp, Result<&2, &2, KErr, Reply>,
470        Conn.read(4294967295n, RChk{s, False{}, False{}, False{}}, H.Http.resp.init(), 60000),
471        p => Pool.run(f, PEv.read(reused, p), cfg, pool, req))
472    case 1n+f PSent{True{}, Fail{e}}:
473      # a stale idle connection: the request was not written, so retry
474      Pool.run(f, PTook{Wait{Unit{}}}, cfg, pool, req)
475    case 1n+f PSent{False{}, Fail{(c, m)}}:
476      IO.pure(Result<&2, &2, KErr, Reply>, Fail{KNet{c, m}})
477    case 1n+f PRead{+reused, p}:
478      Pool.run(f, PEv.fin(reused, Req.idempotent(req), p), cfg, pool, req)
479    case 1n+f PFin{s, +r, True{}, t}:
480      IO.bind(Unit, Result<&2, &2, KErr, Reply>, Pool.put(pool, s),
481        u => IO.pure(Result<&2, &2, KErr, Reply>, Pool.reply(r)))
482    case 1n+f PFin{s, r, False{}, True{}}:
483      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s),
484        u => Pool.run(f, PTook{Wait{Unit{}}}, cfg, pool, req))
485    case 1n+f PFin{s, +r, False{}, False{}}:
486      IO.bind(Unit, Result<&2, &2, KErr, Reply>, N.Tls.close(s),
487        u => IO.pure(Result<&2, &2, KErr, Reply>, Pool.reply(r)))