src/coordinator/OwnerCursorCodec.bend

mylsm-lsm-store@0.6.0.1 · 62 lines · 9 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/coordinator/Protocol.bend as Protocol
3import mylsm-lsm-store@0.6.0.1/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}