cbor
stdlib/extra/cbor.hk: a CBOR (RFC 8949) codec.
CborSerializer / CborDeserializer are pure-Hanki impls of the core Serializer / Deserializer traits (no @intrinsic): they translate the format-agnostic structural events of Encode / Decode to and from the CBOR wire format. @derive(Encode, Decode) on any type therefore yields a CBOR codec for it: encode through a CborSerializer, decode through a CborDeserializer.
Mapping to CBOR major types (RFC 8949 §3): unsigned ints -> major 0 (minimal-length, ignoring the declared fixed width); signed -> major 0/1; bytes -> major 2; string -> major 3 (text); a sequence/struct -> major 4 (array); a map -> major 5; a sum variant -> a major-4 array [tag, ...payloads]; bool/null/f64 -> major 7 (0xf5/0xf4, 0xf6, 0xfb+8). An Option is idiomatic: None is null, Some(v) is just v, and decode peeks for null before reading the payload.
v0 scope: definite-length items only; f64 is read as the 64-bit form (0xfb); decoding foreign half/single floats (0xf9/0xfa) is a tracked follow-up. A wire byte that doesn't match the type the target expects is a BadTag carrying the major type and offset.
CborSerializer
struct CborSerializer
out: BytesBuilder
end
A CBOR serializer over a growable byte sink.
CborDeserializer
struct CborDeserializer
src: BytesReader
end
A CBOR deserializer over a forward byte cursor.
CborHead
struct CborHead
major: u8
value: u64
end
A decoded CBOR item head: its major type (0..7) and argument value.
BytesBuilder
BytesBuilder, or bytes.BytesBuilder, is a runtime-managed native resource handle (HANKI.md §4): live native state the per-actor memory manager owns, released when the last handle drops. It is single-owner - never copied, moved across a send - and has no fields of its own, so its methods are its whole surface. A handle is minted by an API that opens one; it is never constructed.
cborhead!
def _cbor_head!(out: BytesBuilder, major: u8, value: u64) -> ()
An item head: major type in the top 3 bits, then the argument value in the minimal additional-info form (RFC 8949 §3).
cborsigned!
def _cbor_signed!(out: BytesBuilder, n: i64) -> ()
A signed integer: major 0 for non-negative, major 1 (encoding -1-n) for negative.
cborcount!
def _cbor_count!(out: BytesBuilder, major: u8, count: int) -> ()
A non-negative count (i32) as a major-major header.
CBORUINT_MAX
_CBOR_UINT_MAX: int = 18446744073709551615
The largest / smallest int CBOR can hold natively: major 0 spans 0 ..= 2^64-1, major 1 spans -2^64 ..= -1 (it encodes -1-n).
CBORNINT_MIN
_CBOR_NINT_MIN: int = 0 - 18446744073709551616
intclass
def _int_class(n: int) -> u8
0 finite, 1 inf, 2 -inf, 3 undefined. The same classification serializer._int_class makes; repeated because a module's _ names are its own, and because the CBOR spellings below differ from the structural ones. Nested ifs in place of one elif chain: three ==-against-a-constant tests in one chain is H0618, and match has no pattern for a computed value.
intsentinel
def _int_sentinel(class: u8) -> Option<int>
The inverse, for the decode side.
cborsentinel!
def _cbor_sentinel!(out: BytesBuilder, class: u8) -> ()
A sentinel as CBOR's own non-finite value: undefined, or a half-precision infinity. Half and not double: it is the shortest form that says it with no loss, and every CBOR decoder reads it.
intbe_bytes
def _int_be_bytes(m: int) -> List<u8>
The big-endian magnitude bytes of a positive int, for a bignum payload. A division loop per byte: int is unbounded and pure Hanki has no byte view of a bignum. Only reached outside the native 64-bit range, which leaves the cost to the values that need it.
cborbignum!
def _cbor_bignum!(out: BytesBuilder, tag: u64, magnitude: int) -> ()
A bignum: the tag, then the magnitude as a major-2 byte string.
cborint!
def _cbor_int!(out: BytesBuilder, n: int) -> ()
One int, in whichever of the four forms fits it.
impl Serializer<CborSerializer>
put_bool!
def put_bool!(self, b: bool) -> ()
Encodes the boolean as the CBOR simple value true (0xf5) or false (0xf4).
@no-doctest: structural encode op; the round-trip tests below exercise it
put_u8!
def put_u8!(self, n: u8) -> ()
Encodes the byte as a minimal-width CBOR unsigned integer (major type 0).
s = CborSerializer(out=BytesBuilder.new!())
s.put_u8!(200u8)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_u8!()
Ok(v) -> v
Err(_) -> 0u8
end => 200u8
put_u16!
def put_u16!(self, n: u16) -> ()
Encodes the value as a minimal-width CBOR unsigned integer (major type 0).
s = CborSerializer(out=BytesBuilder.new!())
s.put_u16!(4000u16)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_u16!()
Ok(v) -> v
Err(_) -> 0u16
end => 4000u16
put_u32!
def put_u32!(self, n: u32) -> ()
Encodes the value as a minimal-width CBOR unsigned integer (major type 0).
s = CborSerializer(out=BytesBuilder.new!())
s.put_u32!(70000u32)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_u32!()
Ok(v) -> v
Err(_) -> 0u32
end => 70000u32
put_u64!
def put_u64!(self, n: u64) -> ()
Encodes the value as a minimal-width CBOR unsigned integer (major type 0).
s = CborSerializer(out=BytesBuilder.new!())
s.put_u64!(5000000000u64)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_u64!()
Ok(v) -> v
Err(_) -> 0u64
end => 5000000000u64
put_i32!
def put_i32!(self, n: i32) -> ()
Encodes the value as a CBOR integer, widening to the i64 encoder.
s = CborSerializer(out=BytesBuilder.new!())
s.put_i32!(-7i32)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_i32!()
Ok(v) -> v
Err(_) -> 0i32
end => -7i32
put_i64!
def put_i64!(self, n: i64) -> ()
Encode n as a CBOR integer: major type 0 for a non-negative value, major type 1 for a negative one; minimal width in both cases. CborDeserializer.take_i64! reads it back:
s = CborSerializer(out=BytesBuilder.new!())
s.put_i64!(-42i64)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_i64!()
Ok(v) -> v
Err(_) -> 0i64
end => -42i64
put_f64!
def put_f64!(self, x: f64) -> ()
Encodes the number as a CBOR double (0xfb) in IEEE-754 big-endian bits.
s = CborSerializer(out=BytesBuilder.new!())
s.put_f64!(1.5f64)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_f64!()
Ok(v) -> v == 1.5f64
Err(_) -> false
end => true
put_string!
def put_string!(self, s: string) -> ()
Encode s as a CBOR text string (major type 3): a length-prefixed run of its UTF-8 bytes. CborDeserializer.take_string! is the inverse, and a value written through a CborSerializer reads back through a CborDeserializer over the finished bytes:
s = CborSerializer(out=BytesBuilder.new!())
s.put_string!("hi")
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_string!()
Ok(v) -> v
Err(_) -> "<decode failed>"
end => "hi"
put_bytes!
def put_bytes!(self, b: bytes) -> ()
Encodes the buffer as a CBOR byte string (major type 2).
s = CborSerializer(out=BytesBuilder.new!())
s.put_bytes!("hi".to_bytes())
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_bytes!()
Ok(v) -> v == "hi".to_bytes()
Err(_) -> false
end => true
put_none!
def put_none!(self) -> ()
Encodes an absent optional as CBOR null (0xf6).
take_is_some! reads it back as "no payload follows":
s = CborSerializer(out=BytesBuilder.new!())
s.put_none!()
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_is_some!()
Ok(v) -> v
Err(_) -> true
end => false
put_some!
def put_some!(self) -> ()
Some(v) is v in CBOR; the payload's own encode follows.
Nothing is written here: the payload's own encode is the entire representation, and the reader therefore sees a value and no wrapper:
s = CborSerializer(out=BytesBuilder.new!())
s.put_some!()
s.put_u8!(7u8)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
present = match d.take_is_some!()
Ok(b) -> b
Err(_) -> false
end
payload = match d.take_u8!()
Ok(v) -> v
Err(_) -> 0u8
end
present => true
payload => 7u8
begin_seq!
def begin_seq!(self, len: int) -> ()
Open a CBOR array (major type 4) of len items; each element's own encode follows. CborDeserializer.take_seq! reads the count back, then the same number of element decodes follow in order:
s = CborSerializer(out=BytesBuilder.new!())
s.begin_seq!(2)
s.put_i64!(10i64)
s.put_i64!(20i64)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
count = match d.take_seq!()
Ok(c) -> c
Err(_) -> 0
end
first = match d.take_i64!()
Ok(v) -> v
Err(_) -> 0i64
end
count => 2
first => 10i64
begin_map!
def begin_map!(self, len: int) -> ()
Opens a CBOR map (major type 5) of len pairs; each key/value encode follows.
s = CborSerializer(out=BytesBuilder.new!())
s.begin_map!(2)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_map!()
Ok(v) -> v
Err(_) -> 0
end => 2
begin_struct!
def begin_struct!(self, fields: int) -> ()
A struct is a positional CBOR array of its field values.
s = CborSerializer(out=BytesBuilder.new!())
s.begin_struct!(1)
s.put_u8!(9u8)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
field = match d.take_struct!(1)
Ok(()) -> match d.take_u8!()
Ok(v) -> v
Err(_) -> 0u8
end
Err(_) -> 0u8
end
field => 9u8
begin_variant!
def begin_variant!(self, tag: u8, payloads: int) -> ()
A sum variant is the array [tag, ...payloads].
s = CborSerializer(out=BytesBuilder.new!())
s.begin_variant!(3u8, 0)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_variant!()
Ok(v) -> v
Err(_) -> 0u8
end => 3u8
put_int!
def put_int!(self, n: int) -> ()
Encodes the int as a native CBOR integer where one fits, a bignum (tag 2 / tag 3) beyond that, and a sentinel as CBOR's own non-finite value. Overrides the structural default, which writes a class byte and the digits as text: correct, and opaque to a non-Hanki consumer.
s = CborSerializer(out=BytesBuilder.new!())
s.put_int!(10)
s.out.finish!().get(0) => Some(10u8)
put_decimal!
def put_decimal!(self, d: decimal) -> ()
Encodes the decimal as a decimal fraction (tag 4, RFC 8949 §3.4.4): the array [exponent, mantissa], meaning mantissa * 10^exponent. CBOR's exponent counts the other way from Hanki's scale, and is negated. A sentinel has no fraction form and goes out bare.
s = CborSerializer(out=BytesBuilder.new!())
s.put_decimal!(1.5)
b = s.out.finish!()
b.get(0) => Some(196u8)
b.get(1) => Some(130u8)
put_rational!
def put_rational!(self, r: rational) -> ()
Encodes the rational as tag 30 (RFC 8746 §4): the array [numerator, denominator], already reduced with a positive denominator. A sentinel goes out bare, as for decimal.
s = CborSerializer(out=BytesBuilder.new!())
half: rational = 1 / 2
s.put_rational!(half)
b = s.out.finish!()
b.get(0) => Some(216u8)
b.get(1) => Some(30u8)
put_f32!
def put_f32!(self, x: f32) -> ()
Encodes the f32 as a CBOR single-precision float (major 7, ai 26, 0xfa), which is the width the value already has. The structural default would have written the bit pattern as an integer.
s = CborSerializer(out=BytesBuilder.new!())
s.put_f32!(f32.from_bits(1065353216u32))
s.out.finish!().get(0) => Some(250u8)
cborbe_acc
def _cbor_be_acc(b: bytes, i: int, n: int, acc: u64) -> u64
Big-endian reassembly of the first n bytes of b into a u64.
BytesReader
BytesReader, or bytes.BytesReader, is a runtime-managed native resource handle (HANKI.md §4): live native state the per-actor memory manager owns, released when the last handle drops. It is single-owner - never copied, moved across a send - and has no fields of its own, so its methods are its whole surface. A handle is minted by an API that opens one; it is never constructed.
cborarg!
def _cbor_arg!(src: BytesReader, major: u8, n: int) -> Result<CborHead, DecodeError>
cborread_head!
def _cbor_read_head!(src: BytesReader) -> Result<CborHead, DecodeError>
Read one item head: the initial byte's major type and argument value.
cborexpect!
def _cbor_expect!(src: BytesReader, major: u8) -> Result<u64, DecodeError>
Read a head and require its major type, returning the argument value.
cborread_signed!
def _cbor_read_signed!(src: BytesReader) -> Result<i64, DecodeError>
Read a signed integer (major 0 -> n, major 1 -> -1-value).
intfrombebytes
def _int_from_be_bytes(b: bytes) -> int
Big-endian reassembly of a bignum payload into an int. The inverse of _int_be_bytes, and unbounded like it.
cbortake_sentinel!
def _cbor_take_sentinel!(src: BytesReader) -> Result<Option<int>, DecodeError>
Is the next item one of the non-finite forms an exact-tier sentinel takes? If so consume it and answer which; otherwise leave the cursor untouched.
The infinity check covers all three float widths and never the half alone this codec writes: a foreign encoder is free to spell an infinity at any width, and reading it back as inf is the interoperability the tags are here for. Anything else at those heads is a BadTag: a float where an exact-tier value belongs is a wire mismatch and no number to round.
cborinfinity!
def _cbor_infinity!(src: BytesReader, at: int, pos: u64, neg: u64) -> Result<Option<int>, DecodeError>
Consume a float head and map its argument to inf / -inf, given that width's two IEEE infinity patterns.
cbortake_int!
def _cbor_take_int!(src: BytesReader) -> Result<int, DecodeError>
One int in any of the four forms _cbor_int! writes.
cbortake_bignum!
def _cbor_take_bignum!(src: BytesReader, tag: u64, at: int) -> Result<int, DecodeError>
The byte-string payload of a tag-2 / tag-3 bignum, already past its tag.
cbortaketaggedpair!
def _cbor_take_tagged_pair!(src: BytesReader, tag: u64) -> Result<Pair<int, int>, DecodeError>
A tagged 2-element array ([a, b] under tag), the form both tag 4 and tag 30 take. Returns the pair in wire order.
cborread_run!
def _cbor_read_run!(src: BytesReader, major: u8) -> Result<bytes, DecodeError>
Read a length-prefixed run of bytes for a string/bytes item of the given major type.
impl Deserializer<CborDeserializer>
take_bool!
def take_bool!(self) -> Result<bool, DecodeError>
Reads one CBOR boolean; any other tag is a BadTag at its offset.
CBOR spells the two booleans 0xf5 and 0xf4 (RFC 8949 major 7):
d = CborDeserializer(src=BytesReader.new!(245u8.to_bytes()))
match d.take_bool!()
Ok(v) -> v
Err(_) -> false
end => true
take_u8!
def take_u8!(self) -> Result<u8, DecodeError>
Reads a CBOR unsigned integer and narrows it to u8.
s = CborSerializer(out=BytesBuilder.new!())
s.put_u8!(200u8)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_u8!()
Ok(v) -> v
Err(_) -> 0u8
end => 200u8
take_u16!
def take_u16!(self) -> Result<u16, DecodeError>
Reads a CBOR unsigned integer and narrows it to u16.
s = CborSerializer(out=BytesBuilder.new!())
s.put_u16!(4000u16)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_u16!()
Ok(v) -> v
Err(_) -> 0u16
end => 4000u16
take_u32!
def take_u32!(self) -> Result<u32, DecodeError>
Reads a CBOR unsigned integer and narrows it to u32.
s = CborSerializer(out=BytesBuilder.new!())
s.put_u32!(70000u32)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_u32!()
Ok(v) -> v
Err(_) -> 0u32
end => 70000u32
take_u64!
def take_u64!(self) -> Result<u64, DecodeError>
Reads a CBOR unsigned integer (major type 0).
s = CborSerializer(out=BytesBuilder.new!())
s.put_u64!(5000000000u64)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_u64!()
Ok(v) -> v
Err(_) -> 0u64
end => 5000000000u64
take_i32!
def take_i32!(self) -> Result<i32, DecodeError>
Reads a CBOR integer of either sign and narrows it to i32.
s = CborSerializer(out=BytesBuilder.new!())
s.put_i32!(-7i32)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_i32!()
Ok(v) -> v
Err(_) -> 0i32
end => -7i32
take_i64!
def take_i64!(self) -> Result<i64, DecodeError>
Reads a CBOR integer of either sign (major type 0 or 1).
s = CborSerializer(out=BytesBuilder.new!())
s.put_i64!(-42i64)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_i64!()
Ok(v) -> v
Err(_) -> 0i64
end => -42i64
take_f64!
def take_f64!(self) -> Result<f64, DecodeError>
Reads a CBOR double (0xfb); any other tag is a BadTag.
s = CborSerializer(out=BytesBuilder.new!())
s.put_f64!(1.5f64)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_f64!()
Ok(v) -> v == 1.5f64
Err(_) -> false
end => true
take_string!
def take_string!(self) -> Result<string, DecodeError>
Reads a CBOR text string (major type 3), rejecting invalid UTF-8.
s = CborSerializer(out=BytesBuilder.new!())
s.put_string!("hi")
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_string!()
Ok(v) -> v
Err(_) -> "?"
end => "hi"
take_bytes!
def take_bytes!(self) -> Result<bytes, DecodeError>
Reads a CBOR byte string (major type 2).
s = CborSerializer(out=BytesBuilder.new!())
s.put_bytes!("hi".to_bytes())
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_bytes!()
Ok(v) -> v == "hi".to_bytes()
Err(_) -> false
end => true
takeissome!
def take_is_some!(self) -> Result<bool, DecodeError>
None is null (0xf6); anything else is a Some payload left in place.
A null is consumed and answers false; anything else is left in place for the payload's own decode:
d = CborDeserializer(src=BytesReader.new!(246u8.to_bytes()))
match d.take_is_some!()
Ok(v) -> v
Err(_) -> true
end => false
take_seq!
def take_seq!(self) -> Result<int, DecodeError>
Reads an array header (major type 4) and returns its element count.
s = CborSerializer(out=BytesBuilder.new!())
s.begin_seq!(3)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_seq!()
Ok(v) -> v
Err(_) -> 0
end => 3
take_map!
def take_map!(self) -> Result<int, DecodeError>
Reads a map header (major type 5) and returns its pair count.
s = CborSerializer(out=BytesBuilder.new!())
s.begin_map!(2)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_map!()
Ok(v) -> v
Err(_) -> 0
end => 2
take_struct!
def take_struct!(self, fields: int) -> Result<(), DecodeError>
The struct array header; its element count is implied by the type, and the count is read (advancing past the header) but not re-checked here.
s = CborSerializer(out=BytesBuilder.new!())
s.begin_struct!(1)
s.put_u8!(9u8)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
field = match d.take_struct!(1)
Ok(()) -> match d.take_u8!()
Ok(v) -> v
Err(_) -> 0u8
end
Err(_) -> 0u8
end
field => 9u8
take_variant!
def take_variant!(self) -> Result<u8, DecodeError>
The variant array [tag, ...]: read the header, then the tag uint.
s = CborSerializer(out=BytesBuilder.new!())
s.begin_variant!(3u8, 0)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
match d.take_variant!()
Ok(v) -> v
Err(_) -> 0u8
end => 3u8
position!
def position!(self) -> int
The reader's current byte offset, which is what a decode error reports.
s = CborSerializer(out=BytesBuilder.new!())
s.put_u8!(200u8)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
d.position!() => 0
d.take_u8!()
d.position!() => 2
remaining!
def remaining!(self) -> int
How many bytes are left unread in the source buffer.
s = CborSerializer(out=BytesBuilder.new!())
s.put_u8!(200u8)
d = CborDeserializer(src=BytesReader.new!(s.out.finish!()))
d.remaining!() => 2
d.take_u8!()
d.remaining!() => 0
take_int!
def take_int!(self) -> Result<int, DecodeError>
Reads an int from any of the forms put_int! writes: a native CBOR integer, a tag-2 / tag-3 bignum, or one of CBOR's non-finite values for a sentinel.
@no-doctest: structural decode op; the round-trip tests below exercise it
take_decimal!
def take_decimal!(self) -> Result<decimal, DecodeError>
Reads a decimal from the tag-4 decimal fraction put_decimal! writes, negating CBOR's exponent back into Hanki's scale, or from a bare sentinel.
@no-doctest: structural decode op; the round-trip tests below exercise it
take_rational!
def take_rational!(self) -> Result<rational, DecodeError>
Reads a rational from the tag-30 pair put_rational! writes, or from a bare sentinel.
@no-doctest: structural decode op; the round-trip tests below exercise it
take_f32!
def take_f32!(self) -> Result<f32, DecodeError>
Reads an f32 from the CBOR single-precision float put_f32! writes. Only that width: a double would have to be rounded to fit, and nothing in core converts into f32, and a wider float here is a wire mismatch.
@no-doctest: structural decode op; the round-trip tests below exercise it
cborencode_u8
def _cbor_encode_u8(n: u8) -> bytes
cboru64_bytes
def _cbor_u64_bytes(n: u64) -> bytes
cboru64_roundtrip
def _cbor_u64_roundtrip(n: u64) -> u64
_one
def _one(b: u8) -> bytes
A one-byte bytes for golden comparisons (no \xNN string escape).
_written!
def _written!(f: (CborSerializer) -> () [e]) -> bytes [e]
The bytes f writes through a fresh serializer, for golden comparison.
byteat
def _byte_at(b: bytes, i: int) -> u8