1	# Connect client transport (bend-kit-http). See src/README-api.md ("Transport -> Client").
2	#
3	# Every call returns IO(Result<&1, &1, E.Error, Response>):
4	#   - transport failures                    -> Fail(Unavailable, <error text>)
5	#   - unary non-200                         -> Fail(E.from_http(status, body))
6	#   - stream: error in the EndStreamResponse -> Fail(error)  (messages before it are lost;
7	#     use server_stream.each to see them as they arrive)
8	#   - stream without EndStreamResponse      -> Fail(Unknown "missing EndStreamResponse")
9	# Compressed messages are not supported (Fail Internal): no connect-accept-encoding is sent.
10	# The stream readers loop on Http.stream.read, which is @unsafe, so those loops are @unsafe too.
11	import Base
12	import bend-kit-http@0.36.0.0/http.bend as Http
13	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
14	import bend-kit-url@0.4.1.0/url.bend as Url
15	import ../code.bend as Code
16	import ../error.bend as E
17	import ../envelope.bend as Env
18	import ../protocol.bend as P
19	import ../metadata.bend as Md
20	import ../b64.bend as B64
21	
22	type Response is Type:
23	  Response{headers: Map<&2, List<&2, String>>, trailers: Map<&2, List<&2, String>>, messages: List<&1, Bytes.Bytes>}
24	
25	# What a finished stream yields: messages (collect mode) and the EndStream metadata.
26	type Fin is Type:
27	  Fin{messages: List<&1, Bytes.Bytes>, trailers: Map<&2, List<&2, String>>}
28	
29	# ------------------------------------------------------------------ helpers
30	
31	type Item is Type:
32	  Msg{st: Http.Stream, data: Bytes.Bytes, rest: Bytes.Bytes}
33	  Over{error: Maybe<&1, E.Error>, metadata: Map<&2, List<&2, String>>}
34	  Broke{e: E.Error}
35	
36	def max_message() -> U32:
37	  16777216
38	
39	def err.show(e: Http.Err) -> String:
40	  match e:
41	    case Http.ErrUrl{}: "invalid URL"
42	    case Http.ErrDns{}: "DNS resolution failed"
43	    case Http.ErrConnect{code, why}: "connect failed: " ++ why ++ " (" ++ U32.show(code) ++ ")"
44	    case Http.ErrTls{code, why}: "TLS failed: " ++ why
45	    case Http.ErrRead{code, why}: "read failed: " ++ why ++ " (" ++ U32.show(code) ++ ")"
46	    case Http.ErrWrite{code, why}: "write failed: " ++ why ++ " (" ++ U32.show(code) ++ ")"
47	    case Http.ErrTimeout{}: "timeout"
48	    case Http.ErrRedirect{}: "unexpected redirect"
49	    case Http.ErrBad{}: "malformed HTTP response"
50	
51	def unavailable(e: Http.Err) -> E.Error:
52	  E.new(Code.Unavailable{}, err.show(e))
53	
54	def fail(code: Code.Code, msg: String) -> Result<&1, &1, E.Error, Response>:
55	  Fail{E.new(code, msg)}
56	
57	def skip(m: Bytes.Bytes) -> IO(Unit):
58	  IO.pure(Unit, Unit{})
59	
60	def url_of(base_url: String, procedure: String) -> String:
61	  base_url ++ "/" ++ procedure
62	
63	def req_headers(headers: Map<&2, List<&2, String>>, ct: String) -> Map<&2, List<&2, String>>:
64	  Http.set(Http.set(headers, "content-type", ct), "connect-protocol-version", "1")
65	
66	def ct_ok(+headers: Map<&2, List<&2, String>>, expect: String) -> Bool:
67	  String.eq(String.to_lower(String.trim(P.media(Http.header(headers, "content-type")))), expect)
68	
69	# ------------------------------------------------------------------ unary
70	
71	def unary.build(ht: Map<&2, List<&2, String>> & Map<&2, List<&2, String>>, body: Bytes.Bytes) -> Response:
72	  (h, t) = ht
73	  Response{h, t, [body]}
74	
75	def unary.status(ok: Bool, ctok: Bool, +status: U32, headers: Map<&2, List<&2, String>>, body: Bytes.Bytes) -> Result<&1, &1, E.Error, Response>:
76	  match ok:
77	    case False{}:
78	      Fail{E.from_http(status, body)}
79	    case True{}:
80	      match ctok:
81	        case False{}:
82	          fail(Code.Unknown{}, "unexpected response content-type")
83	        case True{}:
84	          Done{unary.build(Md.split_trailers(headers), body)}
85	
86	def unary.res(+codec: P.Codec, res: Http.Res) -> Result<&1, &1, E.Error, Response>:
87	  Http.Res{+status, +headers, body} = res
88	  unary.status(U32.is_eq(status, 200), ct_ok(headers, P.unary_content_type(codec)), status, headers, body)
89	
90	def unary.finish(+codec: P.Codec, r: Result<&1, &1, Http.Err, Http.Res>) -> Result<&1, &1, E.Error, Response>:
91	  match r:
92	    case Fail{e}:
93	      Fail{unavailable(e)}
94	    case Done{res}:
95	      unary.res(codec, res)
96	
97	def unary(base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, msg: Bytes.Bytes) -> IO(Result<&1, &1, E.Error, Response>):
98	  do IO<Result<&1, &1, E.Error, Response>>:
99	    r : Result<&1, &1, Http.Err, Http.Res> <- Http.fetch("POST", url_of(base_url, procedure), req_headers(headers, P.unary_content_type(codec)), msg)
100	    return unary.finish(codec, r)
101	
102	# ------------------------------------------------------------------ GET
103	
104	def get.b64(name: String, msg: Bytes.Bytes) -> String:
105	  "&base64=1&encoding=" ++ Url.pct.encode_q(name) ++ "&message=" ++ B64.encode_url_raw(msg)
106	
107	# Query order follows the Query-Get rule: connect, base64, compression, encoding, message.
108	def get.query(+codec: P.Codec, msg: Bytes.Bytes) -> String:
109	  match codec:
110	    case P.Json{}:
111	      "?connect=v1&encoding=json&message=" ++ Url.pct.encode_q(Bytes.to_string(msg))
112	    case P.Proto{}:
113	      "?connect=v1" ++ get.b64("proto", msg)
114	    case P.Custom{name}:
115	      "?connect=v1" ++ get.b64(name, msg)
116	
117	def get(base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, msg: Bytes.Bytes) -> IO(Result<&1, &1, E.Error, Response>):
118	  do IO<Result<&1, &1, E.Error, Response>>:
119	    r : Result<&1, &1, Http.Err, Http.Res> <- Http.fetch("GET", url_of(base_url, procedure) ++ get.query(codec, msg), headers, Http.from_string(""))
120	    return unary.finish(codec, r)
121	
122	# ------------------------------------------------------------------ streaming reads
123	
124	@unsafe
125	def drain(st: Http.Stream, acc: Bytes.Bytes) -> IO(Bytes.Bytes):
126	  do IO<Bytes.Bytes>:
127	    m : Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>> <- Http.stream.read(st)
128	    drain.step(m, acc)
129	
130	@unsafe
131	def drain.step(m: Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>>, acc: Bytes.Bytes) -> IO(Bytes.Bytes):
132	  (st, r) = m
133	  match r:
134	    case Fail{e}:
135	      do IO<Bytes.Bytes>:
136	        Http.stream.close(st)
137	        return acc
138	    case Done{mb}:
139	      match mb:
140	        case None{}:
141	          do IO<Bytes.Bytes>:
142	            Http.stream.close(st)
143	            return acc
144	        case Some{b}:
145	          drain(st, Bytes.append(acc, b))
146	
147	def emit(~on_msg: Bytes.Bytes -> IO(Unit), collect: Bool, m: Bytes.Bytes, acc: List<&1, Bytes.Bytes>) -> IO(List<&1, Bytes.Bytes>):
148	  match collect:
149	    case True{}:
150	      IO.pure(List<&1, Bytes.Bytes>, m <> acc)
151	    case False{}:
152	      do IO<List<&1, Bytes.Bytes>>:
153	        on_msg(m)
154	        return acc
155	
156	def end.done(error: Maybe<&1, E.Error>, metadata: Map<&2, List<&2, String>>, acc: List<&1, Bytes.Bytes>) -> Result<&1, &1, E.Error, Fin>:
157	  match error:
158	    case Some{e}:
159	      Fail{e}
160	    case None{}:
161	      Done{Fin{List.reverse(&1, Bytes.Bytes, acc), metadata}}
162	
163	def broke(code: Code.Code, msg: String) -> Item:
164	  Broke{E.new(code, msg)}
165	
166	def over.es(es: E.EndStream) -> Item:
167	  E.EndStream{error, metadata} = es
168	  Over{error, metadata}
169	
170	def over.of(m: Maybe<&1, E.EndStream>) -> Item:
171	  match m:
172	    case None{}:
173	      broke(Code.Internal{}, "invalid EndStreamResponse")
174	    case Some{es}:
175	      over.es(es)
176	
177	def recv.last(st: Http.Stream, last: Bool, data: Bytes.Bytes, rest: Bytes.Bytes) -> IO(Item):
178	  match last:
179	    case True{}:
180	      do IO<Item>:
181	        Http.stream.close(st)
182	        return over.of(E.end_decode(data))
183	    case False{}:
184	      IO.pure(Item, Msg{st, data, rest})
185	
186	def recv.cls(st: Http.Stream, compressed: Bool, last: Bool, data: Bytes.Bytes, rest: Bytes.Bytes) -> IO(Item):
187	  match compressed:
188	    case True{}:
189	      do IO<Item>:
190	        Http.stream.close(st)
191	        return broke(Code.Internal{}, "compressed message but no encoding was negotiated")
192	    case False{}:
193	      recv.last(st, last, data, rest)
194	
195	def recv.env(st: Http.Stream, env: Env.Envelope, rest: Bytes.Bytes) -> IO(Item):
196	  Env.Envelope{+flags, data} = env
197	  recv.cls(st, U32.is_ne((flags .&. 1 : U32), 0), U32.is_ne((flags .&. 2 : U32), 0), data, rest)
198	
199	# Reads from the wire until one whole envelope is buffered, then classifies it. The stream
200	# is closed whenever the item is Over or Broke. (Http.stream.read is unsafe, so these are.)
201	@unsafe
202	def recv(st: Http.Stream, buf: Bytes.Bytes) -> IO(Item):
203	  recv.step(st, Env.decode(buf, max_message()))
204	
205	@unsafe
206	def recv.step(st: Http.Stream, step: Env.Step) -> IO(Item):
207	  match step:
208	    case Env.Need{buf}:
209	      do IO<Item>:
210	        m : Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>> <- Http.stream.read(st)
211	        recv.read(m, buf)
212	    case Env.Got{env, rest}:
213	      recv.env(st, env, rest)
214	    case Env.Bad{reason}:
215	      do IO<Item>:
216	        Http.stream.close(st)
217	        return broke(Code.Internal{}, reason)
218	
219	@unsafe
220	def recv.read(m: Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>>, buf: Bytes.Bytes) -> IO(Item):
221	  (st, r) = m
222	  match r:
223	    case Fail{e}:
224	      do IO<Item>:
225	        Http.stream.close(st)
226	        return Broke{unavailable(e)}
227	    case Done{mb}:
228	      match mb:
229	        case None{}:
230	          do IO<Item>:
231	            Http.stream.close(st)
232	            return broke(Code.Unknown{}, "missing EndStreamResponse")
233	        case Some{b}:
234	          recv(st, Bytes.append(buf, b))
235	
236	# The callback loop: one self-recursive template (templates cannot be mutually recursive).
237	@unsafe
238	def pump(~on_msg: Bytes.Bytes -> IO(Unit), +collect: Bool, item: Item, acc: List<&1, Bytes.Bytes>) -> IO(Result<&1, &1, E.Error, Fin>):
239	  match item:
240	    case Msg{st, data, rest}:
241	      do IO<Result<&1, &1, E.Error, Fin>>:
242	        acc2 : List<&1, Bytes.Bytes> <- emit(~on_msg, collect, data, acc)
243	        next : Item <- recv(st, rest)
244	        pump(~on_msg, collect, next, acc2)
245	    case Over{error, metadata}:
246	      IO.pure(Result<&1, &1, E.Error, Fin>, end.done(error, metadata, acc))
247	    case Broke{e}:
248	      IO.pure(Result<&1, &1, E.Error, Fin>, Fail{e})
249	
250	# ------------------------------------------------------------------ response head
251	
252	def done.build(ht: Map<&2, List<&2, String>> & Map<&2, List<&2, String>>, f: Fin) -> Response:
253	  (h, extra) = ht
254	  Fin{messages, trailers} = f
255	  Response{h, trailers, messages}
256	
257	def done(headers: Map<&2, List<&2, String>>, r: Result<&1, &1, E.Error, Fin>) -> Result<&1, &1, E.Error, Response>:
258	  match r:
259	    case Fail{e}:
260	      Fail{e}
261	    case Done{f}:
262	      Done{done.build(Md.split_trailers(headers), f)}
263	
264	def head.ok(~on_msg: Bytes.Bytes -> IO(Unit), collect: Bool, ctok: Bool, st: Http.Stream, headers: Map<&2, List<&2, String>>) -> IO(Result<&1, &1, E.Error, Response>):
265	  match ctok:
266	    case False{}:
267	      do IO<Result<&1, &1, E.Error, Response>>:
268	        Http.stream.close(st)
269	        return fail(Code.Unknown{}, "unexpected response content-type")
270	    case True{}:
271	      do IO<Result<&1, &1, E.Error, Response>>:
272	        first : Item <- recv(st, Bytes.new(0))
273	        r : Result<&1, &1, E.Error, Fin> <- pump(~on_msg, collect, first, Nil{})
274	        return done(headers, r)
275	
276	def head.status(~on_msg: Bytes.Bytes -> IO(Unit), collect: Bool, ok: Bool, ctok: Bool, +status: U32, st: Http.Stream, headers: Map<&2, List<&2, String>>) -> IO(Result<&1, &1, E.Error, Response>):
277	  match ok:
278	    case True{}:
279	      head.ok(~on_msg, collect, ctok, st, headers)
280	    case False{}:
281	      do IO<Result<&1, &1, E.Error, Response>>:
282	        body : Bytes.Bytes <- drain(st, Bytes.new(0))
283	        return Fail{E.from_http(status, body)}
284	
285	def head.of(~on_msg: Bytes.Bytes -> IO(Unit), collect: Bool, expect: String, m: Http.Stream & Http.Res) -> IO(Result<&1, &1, E.Error, Response>):
286	  (st, res) = m
287	  Http.Res{+status, +headers, body} = res
288	  head.status(~on_msg, collect, U32.is_eq(status, 200), ct_ok(headers, expect), status, st, headers)
289	
290	def opened(~on_msg: Bytes.Bytes -> IO(Unit), collect: Bool, +codec: P.Codec, r: Result<&1, &1, Http.Err, Http.Stream>) -> IO(Result<&1, &1, E.Error, Response>):
291	  match r:
292	    case Fail{e}:
293	      IO.pure(Result<&1, &1, E.Error, Response>, Fail{unavailable(e)})
294	    case Done{st}:
295	      head.of(~on_msg, collect, P.stream_content_type(codec), Http.stream.res(st))
296	
297	# ------------------------------------------------------------------ server stream
298	
299	def ss.run(~on_msg: Bytes.Bytes -> IO(Unit), collect: Bool, base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, msg: Bytes.Bytes) -> IO(Result<&1, &1, E.Error, Response>):
300	  do IO<Result<&1, &1, E.Error, Response>>:
301	    r : Result<&1, &1, Http.Err, Http.Stream> <- Http.open("POST", url_of(base_url, procedure), req_headers(headers, P.stream_content_type(codec)), Env.encode(Env.Envelope{0, msg}))
302	    opened(~on_msg, collect, codec, r)
303	
304	def server_stream(base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, msg: Bytes.Bytes) -> IO(Result<&1, &1, E.Error, Response>):
305	  ss.run(~skip, True{}, base_url, procedure, codec, headers, msg)
306	
307	# Like server_stream, but calls on_msg for every message as it arrives (messages in the
308	# returned Response is empty). An error end-stream still yields Fail after the callbacks ran.
309	def server_stream.each(~on_msg: Bytes.Bytes -> IO(Unit), base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, msg: Bytes.Bytes) -> IO(Result<&1, &1, E.Error, Response>):
310	  ss.run(~on_msg, False{}, base_url, procedure, codec, headers, msg)
311	
312	# ------------------------------------------------------------------ client stream
313	
314	@unsafe
315	def upload.next(t: List<&1, Bytes.Bytes>, m: Http.Up & Result<&2, &2, Http.Err, Unit>) -> IO(Http.Up):
316	  (up2, wr) = m
317	  upload.all(up2, t)
318	
319	@unsafe
320	def upload.all(up: Http.Up, msgs: List<&1, Bytes.Bytes>) -> IO(Http.Up):
321	  match msgs:
322	    case Nil{}:
323	      IO.pure(Http.Up, up)
324	    case Con{h, t}:
325	      do IO<Http.Up>:
326	        m : Http.Up & Result<&2, &2, Http.Err, Unit> <- Http.upload.write(up, Env.encode(Env.Envelope{0, h}))
327	        upload.next(t, m)
328	
329	def one(r: Result<&1, &1, E.Error, Response>) -> Result<&1, &1, E.Error, Response>:
330	  match r:
331	    case Fail{e}:
332	      Fail{e}
333	    case Done{resp}:
334	      Response{h, t, ms} = resp
335	      match ms:
336	        case Nil{}:
337	          fail(Code.Unknown{}, "client stream: no response message")
338	        case Con{m, rest}:
339	          match rest:
340	            case Nil{}:
341	              Done{Response{h, t, [m]}}
342	            case Con{x, y}:
343	              fail(Code.Unknown{}, "client stream: more than one response message")
344	
345	def cs.sent(+codec: P.Codec, fr: Result<&1, &1, Http.Err, Http.Stream>) -> IO(Result<&1, &1, E.Error, Response>):
346	  do IO<Result<&1, &1, E.Error, Response>>:
347	    r : Result<&1, &1, E.Error, Response> <- opened(~skip, True{}, codec, fr)
348	    return one(r)
349	
350	def cs.open(+codec: P.Codec, msgs: List<&1, Bytes.Bytes>, r: Result<&1, &1, Http.Err, Http.Up>) -> IO(Result<&1, &1, E.Error, Response>):
351	  match r:
352	    case Fail{e}:
353	      IO.pure(Result<&1, &1, E.Error, Response>, Fail{unavailable(e)})
354	    case Done{up}:
355	      do IO<Result<&1, &1, E.Error, Response>>:
356	        up2 : Http.Up <- upload.all(up, msgs)
357	        fr : Result<&1, &1, Http.Err, Http.Stream> <- Http.upload.finish(up2)
358	        cs.sent(codec, fr)
359	
360	# Each message is one chunked-upload write holding one envelope. A failed write surfaces
361	# from upload.finish.
362	def client_stream(base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, msgs: List<&1, Bytes.Bytes>) -> IO(Result<&1, &1, E.Error, Response>):
363	  do IO<Result<&1, &1, E.Error, Response>>:
364	    r : Result<&1, &1, Http.Err, Http.Up> <- Http.upload("POST", url_of(base_url, procedure), req_headers(headers, P.stream_content_type(codec)))
365	    cs.open(codec, msgs, r)