src/transport/client.bend

bendonnect-rpc@0.1.0.0 · 365 lines · 48 definitions

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.
11import Base
12import bend-kit-http@0.36.0.0/http.bend as Http
13import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
14import bend-kit-url@0.4.1.0/url.bend as Url
15import bendonnect-rpc@0.1.0.0/src/code.bend as Code
16import bendonnect-rpc@0.1.0.0/src/error.bend as E
17import bendonnect-rpc@0.1.0.0/src/envelope.bend as Env
18import bendonnect-rpc@0.1.0.0/src/protocol.bend as P
19import bendonnect-rpc@0.1.0.0/src/metadata.bend as Md
20import bendonnect-rpc@0.1.0.0/src/b64.bend as B64
21
22type 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.
26type Fin is Type:
27  Fin{messages: List<&1, Bytes.Bytes>, trailers: Map<&2, List<&2, String>>}
28
29# ------------------------------------------------------------------ helpers
30
31type 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
36def max_message() -> U32:
37  16777216
38
39def 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
51def unavailable(e: Http.Err) -> E.Error:
52  E.new(Code.Unavailable{}, err.show(e))
53
54def fail(code: Code.Code, msg: String) -> Result<&1, &1, E.Error, Response>:
55  Fail{E.new(code, msg)}
56
57def skip(m: Bytes.Bytes) -> IO(Unit):
58  IO.pure(Unit, Unit{})
59
60def url_of(base_url: String, procedure: String) -> String:
61  base_url ++ "/" ++ procedure
62
63def 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
66def 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
71def 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
75def 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
86def 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
90def 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
97def 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
104def 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.
108def 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
117def 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
125def 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
131def 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
147def 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
156def 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
163def broke(code: Code.Code, msg: String) -> Item:
164  Broke{E.new(code, msg)}
165
166def over.es(es: E.EndStream) -> Item:
167  E.EndStream{error, metadata} = es
168  Over{error, metadata}
169
170def 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
177def 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
186def 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
195def 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
202def recv(st: Http.Stream, buf: Bytes.Bytes) -> IO(Item):
203  recv.step(st, Env.decode(buf, max_message()))
204
205@unsafe
206def 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
220def 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
238def 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
252def 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
257def 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
264def 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
276def 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
285def 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
290def 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
299def 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
304def 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.
309def 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
315def 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
320def 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
329def 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
345def 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
350def 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.
362def 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)