transport.bend
bend-openai-sdk@0.1.0.0 · 90 lines · 0 definitions
1import Base
2import bend-openai-sdk@0.1.0.0/json.bend as J
3
4type Connection is Type:
5 Connection{socket: Socket, buffered: String}
6
7type Read is Type:
8 Frame{connection: Connection, value: J.Json}
9 Broken{message: String}
10
11type Cut is Data:
12 Complete{line: String, rest: String}
13 Partial{}
14
15def cut(s: String, acc: String) -> Cut:
16 match s:
17 case SNil{}: Partial{}
18 case SCon{'\n', rest}: Complete{String.reverse(acc), rest}
19 case SCon{c, rest}: cut(rest, SCon{c, acc})
20
21def broken(socket: Socket, message: String) -> IO(Read):
22 do IO<Read>:
23 Socket.close(socket)
24 return Broken{message}
25
26def decoded(result: Result<&1, &1, String, J.Json>, socket: Socket, rest: String) -> IO(Read):
27 match result:
28 case Fail{error}: broken(socket, "Invalid bridge JSON: " ++ error)
29 case Done{value}: IO.pure(Read, Frame{Connection{socket, rest}, value})
30
31def chunk_nonempty(empty: Bool, socket: Socket, chunk: String, buffered: String,
32 next: Connection -> IO(Read)) -> IO(Read):
33 match empty:
34 case True{}: broken(socket, "Bridge closed before a complete frame")
35 case False{}: next(Connection{socket, buffered ++ chunk})
36
37def chunk(result: Socket & Result<&1, &1, U32 & String, String>, buffered: String,
38 next: Connection -> IO(Read)) -> IO(Read):
39 (socket, received) = result
40 match received:
41 case Fail{(code, message)}: broken(socket, "Bridge receive failed: " ++ message)
42 case Done{+data}: chunk_nonempty(String.is_empty(data), socket, data, buffered, next)
43
44def read_cut(part: Cut, socket: Socket, buffered: String,
45 next: Connection -> IO(Read)) -> IO(Read):
46 match part:
47 case Complete{line, rest}: decoded(J.read(line), socket, rest)
48 case Partial{}:
49 do IO<Read>:
50 received : Socket & Result<&1, &1, U32 & String, String> <- TCP.recv(socket, 4096)
51 chunk(received, buffered, next)
52
53def read_limit(large: Bool, socket: Socket, +buffered: String,
54 next: Connection -> IO(Read)) -> IO(Read):
55 match large:
56 case True{}: broken(socket, "Bridge frame exceeds 4 MiB limit")
57 case False{}: read_cut(cut(buffered, ""), socket, buffered, next)
58
59# Socket progress depends on the external peer, not structural recursion.
60@unsafe
61def read(connection: Connection) -> IO(Read):
62 match connection:
63 case Connection{socket, +buffered}:
64 read_limit((U32.from_nat(String.length(buffered)) > 4198400 : U32), socket, buffered, c => read(c))
65
66def sent(result: Socket & Result<&1, &1, U32 & String, Unit>) -> IO(Result<&1, &1, String, Connection>):
67 (socket, result) = result
68 match result:
69 case Fail{(code, message)}:
70 do IO<Result<&1, &1, String, Connection>>:
71 Socket.close(socket)
72 return Fail{message}
73 case Done{unit}: IO.pure(Result<&1, &1, String, Connection>, Done{Connection{socket, ""}})
74
75def connected(result: Result<&1, &1, U32 & String, Socket>, body: String) -> IO(Result<&1, &1, String, Connection>):
76 match result:
77 case Fail{(code, message)}: IO.pure(Result<&1, &1, String, Connection>, Fail{message})
78 case Done{socket}:
79 do IO<Result<&1, &1, String, Connection>>:
80 result : Socket & Result<&1, &1, U32 & String, Unit> <- TCP.send(socket, body ++ "\n")
81 sent(result)
82
83def connect(port: U32, body: String) -> IO(Result<&1, &1, String, Connection>):
84 do IO<Result<&1, &1, String, Connection>>:
85 result : Result<&1, &1, U32 & String, Socket> <- TCP.connect("127.0.0.1", port)
86 connected(result, body)
87
88def close(connection: Connection) -> IO(Unit):
89 match connection:
90 case Connection{socket, buffered}: Socket.close(socket)