src/transport/rpc.bend

bendonnect-rpc@0.1.0.0 · 242 lines · 34 definitions

1# Typed RPC adapters between generated code and the byte-level transports (Srv, Cli).
2# Reference: docs/runtime.md. Templates come first in every parameter list; the codec functions
3# are the generated `Msg.decode(+codec, b)` / `Msg.encode(+codec, m)`, hence the `@+codec` arrows.
4import Base
5import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
6import bendonnect-rpc@0.1.0.0/src/code.bend as Code
7import bendonnect-rpc@0.1.0.0/src/error.bend as E
8import bendonnect-rpc@0.1.0.0/src/protocol.bend as P
9import bendonnect-rpc@0.1.0.0/src/transport/server.bend as Srv
10import bendonnect-rpc@0.1.0.0/src/transport/client.bend as Cli
11
12# ------------------------------------------------------------------ types
13
14# What a handler learns about the call it serves.
15type Ctx is Data:
16  Ctx{procedure: String, codec: P.Codec, headers: Map<&2, List<&2, String>>}
17
18# A typed server-stream sender: pass it through Rpc.send / the generated per-method send.
19type Out is Type:
20  Out{sink: Srv.Sink, codec: P.Codec}
21
22# ------------------------------------------------------------------ small helpers
23
24def err_reply(e: E.Error) -> Srv.Reply:
25  Srv.Single{Fail{e}, Srv.meta.empty()}
26
27def bad_request(+message: String) -> Srv.Reply:
28  Srv.fail(Code.InvalidArgument{}, message)
29
30def reply_io(r: Srv.Reply) -> IO(Srv.Reply):
31  IO.pure(Srv.Reply, r)
32
33def no_custom(+name: String) -> Srv.Reply:
34  Srv.fail(Code.Unimplemented{}, String.append("codec not supported: ", name))
35
36# Unknown procedure: the server answers 404.
37def no_route(call: Srv.Call) -> IO(Srv.Reply):
38  reply_io(Srv.NoRoute{})
39
40# Bidirectional streaming is not supported by the HTTP/1.1 transport.
41def bidi_unimplemented(call: Srv.Call) -> IO(Srv.Reply):
42  reply_io(Srv.fail(Code.Unimplemented{}, "bidirectional streaming is not supported"))
43
44def route.pick(~hit: Srv.Call -> IO(Srv.Reply), ~miss: Srv.Call -> IO(Srv.Reply), b: Bool, call: Srv.Call) -> IO(Srv.Reply):
45  match b:
46    case True{}:
47      hit(call)
48    case False{}:
49      miss(call)
50
51# if call.procedure == procedure then hit(call) else miss(call); chain these, ending in no_route.
52def route(~hit: Srv.Call -> IO(Srv.Reply), ~miss: Srv.Call -> IO(Srv.Reply), +procedure: String, call: Srv.Call) -> IO(Srv.Reply):
53  Srv.Call{+p, kind, codec, headers, timeout_ms, messages} = call
54  route.pick(~hit, ~miss, String.eq(p, procedure), Srv.Call{p, kind, codec, headers, timeout_ms, messages})
55
56# ------------------------------------------------------------------ server: unary
57
58def enc_res(~Res: Data, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, +codec: P.Codec, r: Result<&1, &1, E.Error, Res>) -> Result<&1, &1, E.Error, Bytes.Bytes>:
59  match r:
60    case Fail{e}:
61      Fail{e}
62    case Done{x}:
63      Done{enc(codec, x)}
64
65def unary.reply(~Res: Data, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, +codec: P.Codec, r: Result<&1, &1, E.Error, Res>) -> Srv.Reply:
66  Srv.Single{enc_res(~Res, ~enc, codec, r), Srv.meta.empty()}
67
68def unary.run(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, ~impl: Ctx -> Req -> IO(Result<&1, &1, E.Error, Res>), ctx: Ctx, +codec: P.Codec, r: Result<&1, &1, E.Error, Req>) -> IO(Srv.Reply):
69  match r:
70    case Fail{e}:
71      reply_io(err_reply(e))
72    case Done{req}:
73      do IO<Srv.Reply>:
74        res : Result<&1, &1, E.Error, Res> <- impl(ctx, req)
75        return unary.reply(~Res, ~enc, codec, res)
76
77def unary.one(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, ~impl: Ctx -> Req -> IO(Result<&1, &1, E.Error, Res>), ctx: Ctx, +codec: P.Codec, ms: List<&1, Bytes.Bytes>) -> IO(Srv.Reply):
78  match ms:
79    case Nil{}:
80      reply_io(bad_request("expected exactly one request message"))
81    case Con{m, rest}:
82      match rest:
83        case Nil{}:
84          unary.run(~Req, ~Res, ~dec, ~enc, ~impl, ctx, codec, dec(codec, m))
85        case Con{x, y}:
86          reply_io(bad_request("expected exactly one request message"))
87
88# Serves a unary method: decode the single request message, run impl, encode the response.
89# A decode error (already an E.Error from the generated decoder) is returned as is.
90def unary(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, ~impl: Ctx -> Req -> IO(Result<&1, &1, E.Error, Res>), call: Srv.Call) -> IO(Srv.Reply):
91  Srv.Call{procedure, kind, +codec, headers, timeout_ms, messages} = call
92  match codec:
93    case P.Custom{name}:
94      reply_io(no_custom(name))
95    case P.Json{}:
96      unary.one(~Req, ~Res, ~dec, ~enc, ~impl, Ctx{procedure, P.Json{}, headers}, P.Json{}, messages)
97    case P.Proto{}:
98      unary.one(~Req, ~Res, ~dec, ~enc, ~impl, Ctx{procedure, P.Proto{}, headers}, P.Proto{}, messages)
99
100# ------------------------------------------------------------------ server: server stream
101
102# Sends one typed message on a server stream.
103def send.go(~T: Data, ~enc: @+codec: P.Codec -> T -> Bytes.Bytes, sink2: IO(Srv.Sink), codec: P.Codec) -> IO(Out):
104  do IO<Out>:
105    s : Srv.Sink <- sink2
106    return Out{s, codec}
107
108def send(~T: Data, ~enc: @+codec: P.Codec -> T -> Bytes.Bytes, out: Out, msg: T) -> IO(Out):
109  Out{sink, +codec} = out
110  send.go(~T, ~enc, Srv.send(sink, enc(codec, msg)), codec)
111
112def ss.fin(r: Out & Result<&1, &1, E.Error, Unit>) -> Srv.Sink & Result<&1, &1, E.Error, Unit>:
113  (out, res) = r
114  Out{sink, codec} = out
115  (sink, res)
116
117def ss.exec(~Req: Data, ~impl: Ctx -> Req -> Out -> IO(Out & Result<&1, &1, E.Error, Unit>), procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, req: Req, sink: Srv.Sink) -> IO(Srv.Sink & Result<&1, &1, E.Error, Unit>):
118  do IO<Srv.Sink & Result<&1, &1, E.Error, Unit>>:
119    r : Out & Result<&1, &1, E.Error, Unit> <- impl(Ctx{procedure, codec, headers}, req, Out{sink, codec})
120    return ss.fin(r)
121
122def ss.reply(~Req: Data, ~impl: Ctx -> Req -> Out -> IO(Out & Result<&1, &1, E.Error, Unit>), procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, r: Result<&1, &1, E.Error, Req>) -> IO(Srv.Reply):
123  match r:
124    case Fail{e}:
125      reply_io(err_reply(e))
126    case Done{req}:
127      reply_io(Srv.Many{sink => ss.exec(~Req, ~impl, procedure, codec, headers, req, sink), Srv.meta.empty()})
128
129def ss.one(~Req: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~impl: Ctx -> Req -> Out -> IO(Out & Result<&1, &1, E.Error, Unit>), procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, ms: List<&1, Bytes.Bytes>) -> IO(Srv.Reply):
130  match ms:
131    case Nil{}:
132      reply_io(bad_request("expected exactly one request message"))
133    case Con{m, rest}:
134      match rest:
135        case Nil{}:
136          ss.reply(~Req, ~impl, procedure, codec, headers, dec(codec, m))
137        case Con{x, y}:
138          reply_io(bad_request("expected exactly one request message"))
139
140# Serves a server-streaming method: impl gets the request and an Out to Rpc.send on, and returns
141# the final Out with Done (stream ends cleanly) or Fail (error in the EndStream message).
142def server_stream(~Req: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~impl: Ctx -> Req -> Out -> IO(Out & Result<&1, &1, E.Error, Unit>), call: Srv.Call) -> IO(Srv.Reply):
143  Srv.Call{procedure, kind, +codec, headers, timeout_ms, messages} = call
144  match codec:
145    case P.Custom{name}:
146      reply_io(no_custom(name))
147    case P.Json{}:
148      ss.one(~Req, ~dec, ~impl, procedure, P.Json{}, headers, messages)
149    case P.Proto{}:
150      ss.one(~Req, ~dec, ~impl, procedure, P.Proto{}, headers, messages)
151
152# ------------------------------------------------------------------ server: client stream
153
154def dec_all(~Req: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, +codec: P.Codec, ms: List<&1, Bytes.Bytes>) -> Result<&1, &1, E.Error, List<&2, Req>>:
155  match ms:
156    case Nil{}:
157      Done{Nil{}}
158    case Con{m, t}:
159      do Result<&1, &1, E.Error, List<&2, Req>>:
160        x : Req <- dec(codec, m)
161        rest : List<&2, Req> <- dec_all(~Req, ~dec, codec, t)
162        return x <> rest
163
164def cs.run(~Req: Data, ~Res: Data, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, ~impl: Ctx -> List<&2, Req> -> IO(Result<&1, &1, E.Error, Res>), ctx: Ctx, +codec: P.Codec, r: Result<&1, &1, E.Error, List<&2, Req>>) -> IO(Srv.Reply):
165  match r:
166    case Fail{e}:
167      reply_io(err_reply(e))
168    case Done{reqs}:
169      do IO<Srv.Reply>:
170        res : Result<&1, &1, E.Error, Res> <- impl(ctx, reqs)
171        return unary.reply(~Res, ~enc, codec, res)
172
173# Serves a client-streaming method: all request messages are decoded first (the transport buffers
174# them), impl answers with the single response.
175def client_stream(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, ~impl: Ctx -> List<&2, Req> -> IO(Result<&1, &1, E.Error, Res>), call: Srv.Call) -> IO(Srv.Reply):
176  Srv.Call{procedure, kind, +codec, headers, timeout_ms, messages} = call
177  match codec:
178    case P.Custom{name}:
179      reply_io(no_custom(name))
180    case P.Json{}:
181      cs.run(~Req, ~Res, ~enc, ~impl, Ctx{procedure, P.Json{}, headers}, P.Json{}, dec_all(~Req, ~dec, P.Json{}, messages))
182    case P.Proto{}:
183      cs.run(~Req, ~Res, ~enc, ~impl, Ctx{procedure, P.Proto{}, headers}, P.Proto{}, dec_all(~Req, ~dec, P.Proto{}, messages))
184
185# ------------------------------------------------------------------ client
186
187def bad_response(+message: String) -> E.Error:
188  E.new(Code.Unknown{}, message)
189
190def one_msg(~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, +codec: P.Codec, ms: List<&1, Bytes.Bytes>) -> Result<&1, &1, E.Error, Res>:
191  match ms:
192    case Nil{}:
193      Fail{bad_response("no response message")}
194    case Con{m, rest}:
195      match rest:
196        case Nil{}:
197          dec(codec, m)
198        case Con{x, y}:
199          Fail{bad_response("more than one response message")}
200
201def one_res(~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, +codec: P.Codec, r: Result<&1, &1, E.Error, Cli.Response>) -> Result<&1, &1, E.Error, Res>:
202  match r:
203    case Fail{e}:
204      Fail{e}
205    case Done{Cli.Response{h, t, ms}}:
206      one_msg(~Res, ~dec, codec, ms)
207
208def dec_msgs(~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, +codec: P.Codec, ms: List<&1, Bytes.Bytes>) -> Result<&1, &1, E.Error, List<&2, Res>>:
209  dec_all(~Res, ~dec, codec, ms)
210
211def many_res(~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, +codec: P.Codec, r: Result<&1, &1, E.Error, Cli.Response>) -> Result<&1, &1, E.Error, List<&2, Res>>:
212  match r:
213    case Fail{e}:
214      Fail{e}
215    case Done{Cli.Response{h, t, ms}}:
216      dec_msgs(~Res, ~dec, codec, ms)
217
218def enc_all(~Req: Data, ~enc: @+codec: P.Codec -> Req -> Bytes.Bytes, +codec: P.Codec, xs: List<&2, Req>) -> List<&1, Bytes.Bytes>:
219  match xs:
220    case Nil{}:
221      Nil{}
222    case Con{x, t}:
223      enc(codec, x) <> enc_all(~Req, ~enc, codec, t)
224
225# Unary call. headers: request headers (lowercase keys). Errors: transport failure -> Unavailable,
226# non-200 -> the server's error, undecodable response -> the decoder's error.
227def call.unary(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, ~enc: @+codec: P.Codec -> Req -> Bytes.Bytes, base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, req: Req) -> IO(Result<&1, &1, E.Error, Res>):
228  do IO<Result<&1, &1, E.Error, Res>>:
229    r : Result<&1, &1, E.Error, Cli.Response> <- Cli.unary(base_url, procedure, codec, headers, enc(codec, req))
230    return one_res(~Res, ~dec, codec, r)
231
232# Server-streaming call; collects every message (an error end-stream turns the whole call into Fail).
233def call.server_stream(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, ~enc: @+codec: P.Codec -> Req -> Bytes.Bytes, base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, req: Req) -> IO(Result<&1, &1, E.Error, List<&2, Res>>):
234  do IO<Result<&1, &1, E.Error, List<&2, Res>>>:
235    r : Result<&1, &1, E.Error, Cli.Response> <- Cli.server_stream(base_url, procedure, codec, headers, enc(codec, req))
236    return many_res(~Res, ~dec, codec, r)
237
238# Client-streaming call: sends all reqs, answers the single response.
239def call.client_stream(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, ~enc: @+codec: P.Codec -> Req -> Bytes.Bytes, base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, reqs: List<&2, Req>) -> IO(Result<&1, &1, E.Error, Res>):
240  do IO<Result<&1, &1, E.Error, Res>>:
241    r : Result<&1, &1, E.Error, Cli.Response> <- Cli.client_stream(base_url, procedure, codec, headers, enc_all(~Req, ~enc, codec, reqs))
242    return one_res(~Res, ~dec, codec, r)