1	import Base
2	import ./json.bend as J
3	import ./utf8.bend as U
4	import ./http.bend as H
5	import ./net.bend as N
6	import ./core.bend as C
7	import ./config.bend as K
8	import ./client.bend as Cl
9	
10	# Stream
11	# ======
12	#
13	# A GET whose body arrives over time, such as a pod's log with
14	# follow=true: each cut of body bytes is folded into a state by a callback
15	# as it is decoded (chunked, sized, or read until close), until the body
16	# ends. A status other than 200 is a KApi error carrying the body's Status
17	# message. The connection is the stream's own, never pooled.
18	
19	def Res(-S: Type) -> Type:
20	  Result<&1, &1, Cl.KErr, S>
21	
22	# What the loop just saw: bytes (or a failure) from a recv, or the decoder
23	# after them with the body bytes it decoded, whether the server failed or
24	# the stream is over, whether the answer is 200, and whether its head came.
25	type SEv is Type:
26	  SGot{sock: Socket, resp: H.Resp, r: Result<&1, &1, U32 & String, List<&2, U32>>}
27	  SChk{sock: Socket, resp: H.Resp, body: List<&2, U32>, err: Maybe<&2, String>, ok: Bool,
28	    headed: Bool, empty: Bool, fin: Bool, code: U32}
29	
30	def SEv.got(p: Socket & Result<&1, &1, U32 & String, List<&2, U32>>, resp: H.Resp) -> SEv:
31	  match p:
32	    case (s, r):
33	      SGot{s, resp, r}
34	
35	def Stream.code(r: H.Resp) -> U32:
36	  match r:
37	    case H.Resp{p, Some{H.Head{c, reason, hs}}, out}:
38	      c
39	    case _:
40	      0
41	
42	def Stream.check(s: Socket, +r: H.Resp, +body: List<&2, U32>, eof: Bool) -> SEv:
43	  SChk{s, r, body, H.Http.error(r), U32.is_eq(Stream.code(r), 200),
44	    Bool.not(U32.is_zero(Stream.code(r))), List.is_empty(&2, U32, body),
45	    Bool.or(eof, H.Http.is_done(r)), Stream.code(r)}
46	
47	def Stream.taken(s: Socket, eof: Bool, rb: H.Resp & List<&2, U32>) -> SEv:
48	  match rb:
49	    case (+r, +body):
50	      Stream.check(s, r, body, eof)
51	
52	def Stream.ended(eof: Bool, r: H.Resp) -> H.Resp:
53	  match eof:
54	    case True{}:
55	      H.Http.eof(r)
56	    case False{}:
57	      r
58	
59	# Feeds a cut (empty at the end of the stream) and takes the body bytes it
60	# decoded.
61	def Stream.fed(s: Socket, +bs: List<&2, U32>, r: H.Resp) -> SEv:
62	  Stream.taken(s, List.is_empty(&2, U32, bs),
63	    H.Http.take(Stream.ended(List.is_empty(&2, U32, bs), H.Http.feed(bs, r))))
64	
65	def Stream.message(+text: String, j: Result<&2, &2, String, J.Json>) -> String:
66	  match j:
67	    case Done{s}:
68	      J.Json.str_or(J.Json.at(s, ["message"]), text)
69	    case Fail{m}:
70	      text
71	
72	def Stream.refused(+code: U32, +bad: List<&2, U32>) -> Cl.KErr:
73	  Cl.KApi{C.Status{code, "",
74	    Stream.message(String.trim(U.Utf8.decode_all(bad)), J.Json.parse(U.Utf8.decode_all(bad)))}}
75	
76	def Stream.follow(on: Bool) -> String:
77	  match on:
78	    case True{}:
79	      "&follow=true"
80	    case False{}:
81	      ""
82	
83	def Stream.recv(s: Socket, +ms: U32) ->
84	  IO(Socket & Result<&1, &1, U32 & String, List<&2, U32>>):
85	  N.Tls.recv(s, 65536, ms)
86	
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.
89	def 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))
122	
123	def Stream.close(~S: Type, p: Socket & Res(S)) -> IO(Res(S)):
124	  match p:
125	    case (s, r):
126	      IO.bind(Unit, Res(S), N.Tls.close(s), u => IO.pure(Res(S), r))
127	
128	def Stream.opened(~S: Type, ~on: S -> List<&2, U32> -> IO(S), st: S,
129	  t: Result<&1, &1, U32 & String, N.Tls>, +ms: U32) -> IO(Res(S)):
130	  match t:
131	    case Done{N.Tls{s}}:
132	      IO.bind(Socket & Res(S), Res(S),
133	        Stream.loop(~S, ~on, 4294967295n,
134	          SChk{s, H.Http.resp.init(), Nil{}, None{}, False{}, False{}, True{}, False{}, 0}, st,
135	          Nil{}, ms), p => Stream.close(~S, p))
136	    case Fail{(c, m)}:
137	      IO.pure(Res(S), Fail{Cl.KNet{c, m}})
138	
139	# GETs path (with its query) and folds each cut of the body into st with
140	# on, in order, until the body ends: Done with the last state. ms bounds
141	# each wait for bytes (0: none; a followed log may stay quiet for long).
142	def Stream.fold(~S: Type, ~on: S -> List<&2, U32> -> IO(S), st: S, +cfg: K.Config, path: String,
143	  +ms: U32) -> IO(Res(S)):
144	  IO.bind(Result<&1, &1, U32 & String, N.Tls>, Res(S),
145	    Cl.Conn.open(cfg, H.Req{"GET", path, Nil{}, ""}, False{}), t => Stream.opened(~S, ~on, st, t, ms))
146	
147	# The log of a pod's container, followed while it runs (follow=true): the
148	# stream ends when the container does.
149	def Stream.log_path(ns: String, pod: String, container: String, follow: Bool) -> String:
150	  "/api/v1/namespaces/" ++ H.Http.encode(ns) ++ "/pods/" ++ H.Http.encode(pod) ++
151	    "/log?container=" ++ H.Http.encode(container) ++ Stream.follow(follow)