stream.bend
bend-kit-stream@0.2.0.1 · 173 lines · 0 definitions
1# Bounded packed-byte transfers between files and TCP/TLS sockets.
2import Base
3import bend-kit-files@0.2.0.0/files.bend as Files
4import bend-kit-wire@0.4.6.2/wire.bend as Wire
5
6type Error is Data:
7 BadChunk{}
8 Limit{}
9 Read{code: U32, message: String}
10 Write{code: U32, message: String}
11
12type Outcome is Data:
13 Complete{count: U32}
14 Failed{count: U32, error: Error}
15
16# Internal pure decisions, separated from the trusted effects.
17type ReadAction is Data:
18 End{}
19 Over{}
20 Forward{}
21
22def chunk.valid(+chunk: U32) -> Bool:
23 U32.is_ge(chunk, 1) && U32.is_le(chunk, 1048576)
24
25# The one-byte probe is only requested after the entire cap was forwarded.
26def request(+chunk: U32, +remaining: U32) -> U32:
27 match remaining:
28 case 0:
29 1
30 case _:
31 U32.min(chunk, remaining)
32
33def read.action.flags(zero: Bool, probe: Bool) -> ReadAction:
34 match zero:
35 case True{}:
36 End{}
37 case False{}:
38 match probe:
39 case True{}:
40 Over{}
41 case False{}:
42 Forward{}
43
44def read.action(len: U32, remaining: U32) -> ReadAction:
45 read.action.flags(U32.is_zero(len), U32.is_zero(remaining))
46
47type Transfer<-S: Type, -D: Type> is Type:
48 More{source: S, destination: D, count: U32}
49 Stop{source: S, destination: D, outcome: Outcome}
50
51def file.read(source: File, max: U32, tls: Bool, ms: U32) ->
52 IO(File & Result<&1, &1, U32 & String, U32 & Array<U32>>):
53 Files.read.words(source, max)
54
55def file.write(destination: File, len: U32, words: Array<U32>, tls: Bool) ->
56 IO(File & Result<&1, &1, U32 & String, Unit>):
57 Files.write.words(destination, len, words)
58
59def socket.read.tls(tls: Bool, source: Socket, max: U32, ms: U32) ->
60 IO(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>):
61 match tls:
62 case True{}:
63 Wire.tls.recv.words(source, max, ms)
64 case False{}:
65 Wire.recv.words(source, max, ms)
66
67def socket.read(source: Socket, max: U32, tls: Bool, ms: U32) ->
68 IO(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>):
69 socket.read.tls(tls, source, max, ms)
70
71def socket.write.tls(tls: Bool, destination: Socket, len: U32, words: Array<U32>) ->
72 IO(Socket & Result<&1, &1, U32 & String, Unit>):
73 match tls:
74 case True{}:
75 Wire.tls.send.words(destination, len, words)
76 case False{}:
77 Wire.send.words(destination, len, words)
78
79def socket.write(destination: Socket, len: U32, words: Array<U32>, tls: Bool) ->
80 IO(Socket & Result<&1, &1, U32 & String, Unit>):
81 socket.write.tls(tls, destination, len, words)
82
83def transfer.written(~S: Type, ~D: Type, source: S, count: U32, len: U32,
84 pair: D & Result<&1, &1, U32 & String, Unit>) -> Transfer<S, D>:
85 (destination, result) = pair
86 match result:
87 case Fail{(code, message)}:
88 Stop{source, destination, Failed{count, Write{code, message}}}
89 case Done{unit}:
90 More{source, destination, (count + len : U32)}
91
92def transfer.action(~S: Type, ~D: Type,
93 ~write: D -> U32 -> Array<U32> -> Bool -> IO(D & Result<&1, &1, U32 & String, Unit>),
94 action: ReadAction, source: S, destination: D, count: U32, +len: U32,
95 words: Array<U32>, tls: Bool) -> IO(Transfer<S, D>):
96 match action:
97 case End{}:
98 IO.pure(Transfer<S, D>, Stop{source, destination, Complete{count}})
99 case Over{}:
100 IO.pure(Transfer<S, D>, Stop{source, destination, Failed{count, Limit{}}})
101 case Forward{}:
102 do IO<Transfer<S, D>>:
103 pair : D & Result<&1, &1, U32 & String, Unit> <- write(destination, len, words, tls)
104 return transfer.written(~S, ~D, source, count, len, pair)
105
106def transfer.read(~S: Type, ~D: Type,
107 ~write: D -> U32 -> Array<U32> -> Bool -> IO(D & Result<&1, &1, U32 & String, Unit>),
108 pair: S & Result<&1, &1, U32 & String, U32 & Array<U32>>,
109 destination: D, count: U32, remaining: U32, tls: Bool) -> IO(Transfer<S, D>):
110 (source, result) = pair
111 match result:
112 case Fail{(code, message)}:
113 IO.pure(Transfer<S, D>, Stop{source, destination, Failed{count, Read{code, message}}})
114 case Done{(+len, words)}:
115 transfer.action(~S, ~D, ~write, read.action(len, remaining),
116 source, destination, count, len, words, tls)
117
118# cap+1 reads suffice even if every successful nonempty read contains one byte.
119# Fuel is first among runtime parameters; no fixed read ceiling or unsafe recursion.
120def transfer.loop(~S: Type, ~D: Type,
121 ~read: S -> U32 -> Bool -> U32 -> IO(S & Result<&1, &1, U32 & String, U32 & Array<U32>>),
122 ~write: D -> U32 -> Array<U32> -> Bool -> IO(D & Result<&1, &1, U32 & String, Unit>),
123 fuel: Nat, state: Transfer<S, D>, +tls: Bool, +chunk: U32, +cap: U32, +ms: U32) ->
124 IO(S & D & Outcome):
125 match fuel:
126 case 0n:
127 match state:
128 case Stop{source, destination, outcome}:
129 IO.pure(S & D & Outcome, (source, destination, outcome))
130 case More{source, destination, count}:
131 # Unreachable under the read.words bounds and positive-progress contract.
132 IO.pure(S & D & Outcome, (source, destination, Failed{count, Limit{}}))
133 case 1n+rest:
134 match state:
135 case Stop{source, destination, outcome}:
136 IO.pure(S & D & Outcome, (source, destination, outcome))
137 case More{source, destination, +count}:
138 +remaining = (cap - count : U32)
139 do IO<S & D & Outcome>:
140 pair : S & Result<&1, &1, U32 & String, U32 & Array<U32>> <-
141 read(source, request(chunk, remaining), tls, ms)
142 next : Transfer<S, D> <- transfer.read(~S, ~D, ~write,
143 pair, destination, count, remaining, tls)
144 transfer.loop(~S, ~D, ~read, ~write, rest, next, tls, chunk, cap, ms)
145
146def transfer.start(~S: Type, ~D: Type,
147 ~read: S -> U32 -> Bool -> U32 -> IO(S & Result<&1, &1, U32 & String, U32 & Array<U32>>),
148 ~write: D -> U32 -> Array<U32> -> Bool -> IO(D & Result<&1, &1, U32 & String, Unit>),
149 valid: Bool, source: S, destination: D, tls: Bool, chunk: U32, +cap: U32, ms: U32) ->
150 IO(S & D & Outcome):
151 match valid:
152 case False{}:
153 IO.pure(S & D & Outcome, (source, destination, Failed{0, BadChunk{}}))
154 case True{}:
155 transfer.loop(~S, ~D, ~read, ~write, 1n+U32.to_nat(cap),
156 More{source, destination, 0}, tls, chunk, cap, ms)
157
158# Return both handles on every outcome; never close or retry a failed chunk.
159def file.file(source: File, destination: File, +chunk: U32, cap: U32) ->
160 IO(File & File & Outcome):
161 transfer.start(~File, ~File, ~file.read, ~file.write,
162 chunk.valid(chunk), source, destination, False{}, chunk, cap, 0)
163
164def file.socket(source: File, destination: Socket, tls: Bool, +chunk: U32, cap: U32) ->
165 IO(File & Socket & Outcome):
166 transfer.start(~File, ~Socket, ~file.read, ~socket.write,
167 chunk.valid(chunk), source, destination, tls, chunk, cap, 0)
168
169# ms is the unchanged per-read Wire deadline, including the one-byte cap probe.
170def socket.file(source: Socket, destination: File, tls: Bool, +chunk: U32, cap: U32, ms: U32) ->
171 IO(Socket & File & Outcome):
172 transfer.start(~Socket, ~File, ~socket.read, ~file.write,
173 chunk.valid(chunk), source, destination, tls, chunk, cap, ms)