src/coordinator/OwnerCursorCodec.bend
mylsm-lsm-store@0.6.0.0 · 62 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/coordinator/Protocol.bend as Protocol
3import mylsm-lsm-store@0.6.0.0/src/coordinator/CursorCodec.bend as CursorCodec
4import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
5
6# Reports one owner-issued cursor identifier or a rejected open response.
7type OpenDecision is Type:
8 Opened{cursor_id: U32}
9 OpenRejected{status: Protocol.ResponseStatus}
10 OpenMalformed{}
11
12# Reports one decoded cursor frame or a rejected next response.
13type NextDecision is Type:
14 NextFrame{frame: CursorCodec.Frame}
15 NextRejected{status: Protocol.ResponseStatus}
16 NextMalformed{}
17
18# Reports whether an owner acknowledged cursor closure with an exact body.
19type CloseDecision is Data:
20 CloseSucceeded{}
21 CloseRejected{status: Protocol.ResponseStatus}
22 CloseMalformed{}
23
24# Decodes the owner cursor ID only from a successful exact token payload.
25# Converts an exact token parse into a typed cursor-open decision.
26def open.parsed(parsed: Maybe<&2, U32>) -> OpenDecision:
27 match parsed:
28 case Some{cursor_id}: Opened{cursor_id}
29 case None{}: OpenMalformed{}
30
31# Decodes a successful owner cursor-open response or preserves its rejection.
32def open(status: Protocol.ResponseStatus, payload: Bytes.Bytes) -> OpenDecision:
33 match status:
34 case Protocol.Success{}:
35 open.parsed(Protocol.snapshot.id.read(payload))
36 case _: OpenRejected{status}
37
38# Converts one exact cursor frame decode into a typed client decision.
39def next.decoded(decision: CursorCodec.Decision) -> NextDecision:
40 match decision:
41 case CursorCodec.Decoded{frame}: NextFrame{frame}
42 case CursorCodec.Malformed{}: NextMalformed{}
43
44# Decodes a raw-byte cursor frame only from a successful owner response.
45def next(status: Protocol.ResponseStatus, payload: Bytes.Bytes) -> NextDecision:
46 match status:
47 case Protocol.Success{}:
48 next.decoded(CursorCodec.decode(payload))
49 case _: NextRejected{status}
50
51# Accepts one close payload only when its measured length is zero.
52def close.measured(measured: Bytes.Bytes & U32) -> CloseDecision:
53 match measured:
54 case (_, 0): CloseSucceeded{}
55 case _: CloseMalformed{}
56
57# Accepts cursor close only when success carries no trailing payload.
58def close(status: Protocol.ResponseStatus, payload: Bytes.Bytes) -> CloseDecision:
59 match status:
60 case Protocol.Success{}:
61 close.measured(Bytes.length(payload))
62 case _: CloseRejected{status}