///|
/// Failures produced by Avro Single-Object Encoding envelopes.
pub(all) suberror SingleObjectError {
InvalidMagic
TruncatedHeader(actual~ : Int, required~ : Int)
UnknownFingerprint(UInt64)
FingerprintMismatch(expected~ : UInt64, actual~ : UInt64)
Datum(@codec.CodecError)
} derive(Debug, Eq)
///|
/// Return the two-byte Avro Single-Object Encoding marker (`C3 01`).
pub fn magic() -> Bytes {
[0xc3, 0x01]
}
///|
/// Read a little-endian CRC-64-AVRO fingerprint from the eight bytes following
/// the single-object marker. The caller has already checked the input length.
fn read_fingerprint_unchecked(input : Bytes) -> UInt64 {
let mut result = 0UL
for index in 0..<8 {
result = result | (input[index + 2].to_uint64() << (index * 8))
}
result
}
///|
/// Validate the ten-byte envelope prefix and return its schema fingerprint.
pub fn fingerprint(input : Bytes) -> UInt64 raise SingleObjectError {
if input.length() < 2 {
raise TruncatedHeader(actual=input.length(), required=2)
}
if input[0] != 0xc3 || input[1] != 0x01 {
raise InvalidMagic
}
if input.length() < 10 {
raise TruncatedHeader(actual=input.length(), required=10)
}
read_fingerprint_unchecked(input)
}
///|
/// Encode one datum using Avro Single-Object Encoding. The result contains the
/// marker, the schema's CRC-64-AVRO fingerprint in little-endian order, and one
/// ordinary Avro binary datum payload.
pub fn encode(
schema : @schema.Schema,
datum : @codec.Datum,
) -> Bytes raise SingleObjectError {
let payload = @codec.encode(schema, datum) catch {
error => raise Datum(error)
}
let output = @codec.Encoder::new(
limits=@codec.CodecLimits::new(max_bytes=10 + payload.length()),
)
output.write_raw(magic()) catch {
error => raise Datum(error)
}
output.write_raw(schema.fingerprint64().to_le_bytes()) catch {
error => raise Datum(error)
}
output.write_raw(payload) catch {
error => raise Datum(error)
}
output.to_bytes()
}
///|
/// Decode a Single-Object Encoding value with an expected schema. A mismatch
/// is reported before the datum payload is interpreted.
pub fn decode(
schema : @schema.Schema,
input : Bytes,
limits? : @codec.CodecLimits = @codec.CodecLimits::new(),
) -> @codec.Datum raise SingleObjectError {
let actual = fingerprint(input)
let expected = schema.fingerprint64()
if actual != expected {
raise FingerprintMismatch(expected~, actual~)
}
let payload = input[10:].to_owned()
@codec.decode(schema, payload, limits~) catch {
error => raise Datum(error)
}
}
///|
/// Resolve the writer schema by fingerprint and decode one Single-Object
/// Encoding value. The resolver can use a registry, generated table or local
/// map without coupling this package to a particular storage mechanism.
pub fn decode_resolved(
input : Bytes,
resolve : (UInt64) -> @schema.Schema?,
limits? : @codec.CodecLimits = @codec.CodecLimits::new(),
) -> (@schema.Schema, @codec.Datum) raise SingleObjectError {
let writer_fingerprint = fingerprint(input)
let schema = match resolve(writer_fingerprint) {
Some(schema) => schema
None => raise UnknownFingerprint(writer_fingerprint)
}
let payload = input[10:].to_owned()
let datum = @codec.decode(schema, payload, limits~) catch {
error => raise Datum(error)
}
(schema, datum)
}
///|
/// Return whether an encoded value carries the supplied schema fingerprint.
/// Malformed and truncated envelopes return `false` rather than raising.
pub fn matches_schema(input : Bytes, schema : @schema.Schema) -> Bool {
let value = fingerprint(input) catch { _ => return false }
value == schema.fingerprint64()
}