///|
fn mismatch(schema : @schema.Schema, datum : Datum) -> Unit raise CodecError {
raise TypeMismatch(expected=schema.kind_name(), actual=datum.type_name())
}
///|
fn checked_depth(decoder : Decoder, depth : Int) -> Unit raise CodecError {
if depth >= decoder.limits().max_depth() {
raise LimitExceeded(
offset=decoder.offset(),
limit="maximum schema nesting depth",
)
}
}
///|
fn write_boolean(encoder : Encoder, value : Bool) -> Unit raise CodecError {
encoder.write_byte(if value { 1 } else { 0 })
}
///|
fn read_boolean(decoder : Decoder) -> Bool raise CodecError {
let offset = decoder.offset()
match decoder.read_byte() {
0 => false
1 => true
_ => raise InvalidEncoding(offset~, message="boolean byte must be 0 or 1")
}
}
///|
fn record_value(fields : Array[(String, Datum)], name : String) -> Datum? {
for _, pair in fields {
let (field_name, value) = pair
if field_name == name {
return Some(value)
}
}
None
}
///|
fn is_record_field(fields : Array[@schema.Field], name : String) -> Bool {
for field in fields {
if field.name() == name {
return true
}
}
false
}
///|
fn checked_collection_count(
decoder : Decoder,
count : Int64,
) -> Int raise CodecError {
let offset = decoder.offset()
if count < 0 ||
count > Int64::from_int(decoder.limits().max_collection_items()) {
raise InvalidLength(offset~, length=count)
}
count.to_int()
}
///|
fn collection_block(
decoder : Decoder,
count : Int64,
) -> (Int, Int?) raise CodecError {
if count >= 0 {
(checked_collection_count(decoder, count), None)
} else {
if count == 0x8000000000000000UL.reinterpret_as_int64() {
raise InvalidLength(offset=decoder.offset(), length=count)
}
let item_count = checked_collection_count(decoder, -count)
let size_offset = decoder.offset()
let declared_size = decoder.read_long()
if declared_size < 0 ||
declared_size > Int64::from_int(decoder.limits().max_bytes()) {
raise InvalidLength(offset=size_offset, length=declared_size)
}
(item_count, Some(declared_size.to_int()))
}
}
///|
fn check_block_size(
decoder : Decoder,
start : Int,
declared_size : Int?,
) -> Unit raise CodecError {
match declared_size {
Some(size) =>
if decoder.offset() - start != size {
raise InvalidEncoding(
offset=start,
message="collection block size does not match its payload",
)
}
None => ()
}
}
///|
#warnings("-unnecessary_annotation")
fn write_schema(
root : @schema.Schema,
schema : @schema.Schema,
datum : Datum,
encoder : Encoder,
depth : Int,
) -> Unit raise CodecError {
if depth >= encoder.limits.max_depth() {
raise LimitExceeded(
offset=encoder.length(),
limit="maximum schema nesting depth",
)
}
match schema.kind() {
@schema.SchemaKind::Primitive(@schema.Primitive::Null) =>
match datum {
Null => ()
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Primitive(@schema.Primitive::Boolean) =>
match datum {
Boolean(value) => write_boolean(encoder, value)
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Primitive(@schema.Primitive::Int) =>
match datum {
Int(value) => encoder.write_int(value)
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Primitive(@schema.Primitive::Long) =>
match datum {
Long(value) => encoder.write_long(value)
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Primitive(@schema.Primitive::Float) =>
match datum {
Float(value) => encoder.write_float(value)
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Primitive(@schema.Primitive::Double) =>
match datum {
Double(value) => encoder.write_double(value)
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Primitive(@schema.Primitive::Bytes) =>
match datum {
Bytes(value) => encoder.write_bytes(value)
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Primitive(@schema.Primitive::String) =>
match datum {
String(value) => encoder.write_string(value)
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Record(name=_, namespace_=_, aliases=_, fields~) =>
match datum {
Record(values) => {
for field in fields {
match record_value(values, field.name()) {
Some(value) =>
write_schema(root, field.schema(), value, encoder, depth + 1)
None => raise MissingField(field=field.name())
}
}
for _, pair in values {
let (field_name, _) = pair
if !is_record_field(fields, field_name) {
raise InvalidEncoding(
offset=encoder.length(),
message="record contains a field absent from its schema: \{field_name}",
)
}
}
}
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Enum(
name=_,
namespace_=_,
aliases=_,
symbols~,
default_symbol=_
) =>
match datum {
Enum(index, symbol) => {
if index < 0 || index >= symbols.length() || symbols[index] != symbol {
raise InvalidEncoding(
offset=encoder.length(),
message="enum index and symbol do not match its schema",
)
}
encoder.write_int(index)
}
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Array(items~) =>
match datum {
Array(values) => {
if values.length() > encoder.limits.max_collection_items() {
raise LimitExceeded(
offset=encoder.length(),
limit="maximum collection item count",
)
}
if !values.is_empty() {
encoder.write_long(Int64::from_int(values.length()))
for value in values {
write_schema(root, items, value, encoder, depth + 1)
}
}
encoder.write_long(0L)
}
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Map(values~) =>
match datum {
Map(entries) => {
if entries.length() > encoder.limits.max_collection_items() {
raise LimitExceeded(
offset=encoder.length(),
limit="maximum collection item count",
)
}
if !entries.is_empty() {
encoder.write_long(Int64::from_int(entries.length()))
for key, value in entries {
encoder.write_string(key)
write_schema(root, values, value, encoder, depth + 1)
}
}
encoder.write_long(0L)
}
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Union(branches~) =>
match datum {
Union(index, value) => {
if index < 0 || index >= branches.length() {
raise InvalidEncoding(
offset=encoder.length(),
message="union branch index is outside schema",
)
}
encoder.write_long(Int64::from_int(index))
write_schema(root, branches[index], value, encoder, depth + 1)
}
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Fixed(name=_, namespace_=_, aliases=_, size~) =>
match datum {
Fixed(value) => {
if value.length() != size {
raise InvalidEncoding(
offset=encoder.length(),
message="fixed datum has the wrong size",
)
}
encoder.write_raw(value)
}
_ => mismatch(schema, datum)
}
@schema.SchemaKind::Named(name) =>
match root.resolve_named(name) {
Some(kind) =>
write_schema(
root,
@schema.Schema::new(kind, name),
datum,
encoder,
depth + 1,
)
None =>
raise InvalidEncoding(
offset=encoder.length(),
message="unresolved named schema: \{name}",
)
}
}
}
///|
#warnings("-unnecessary_annotation")
fn read_schema(
root : @schema.Schema,
schema : @schema.Schema,
decoder : Decoder,
depth : Int,
) -> Datum raise CodecError {
checked_depth(decoder, depth)
match schema.kind() {
@schema.SchemaKind::Primitive(@schema.Primitive::Null) => Null
@schema.SchemaKind::Primitive(@schema.Primitive::Boolean) =>
Boolean(read_boolean(decoder))
@schema.SchemaKind::Primitive(@schema.Primitive::Int) =>
Int(decoder.read_int())
@schema.SchemaKind::Primitive(@schema.Primitive::Long) =>
Long(decoder.read_long())
@schema.SchemaKind::Primitive(@schema.Primitive::Float) =>
Float(decoder.read_float())
@schema.SchemaKind::Primitive(@schema.Primitive::Double) =>
Double(decoder.read_double())
@schema.SchemaKind::Primitive(@schema.Primitive::Bytes) =>
Bytes(decoder.read_bytes())
@schema.SchemaKind::Primitive(@schema.Primitive::String) =>
String(decoder.read_string())
@schema.SchemaKind::Record(name=_, namespace_=_, aliases=_, fields~) => {
let values : Array[(String, Datum)] = []
for field in fields {
values.push(
(field.name(), read_schema(root, field.schema(), decoder, depth + 1)),
)
}
Record(values)
}
@schema.SchemaKind::Enum(
name=_,
namespace_=_,
aliases=_,
symbols~,
default_symbol=_
) => {
let offset = decoder.offset()
let index = decoder.read_int()
if index < 0 || index >= symbols.length() {
raise InvalidEncoding(offset~, message="enum index is outside schema")
}
Enum(index, symbols[index])
}
@schema.SchemaKind::Array(items~) => {
let result : Array[Datum] = []
let mut count = decoder.read_long()
while count != 0L {
let (item_count, declared_size) = collection_block(decoder, count)
if item_count >
decoder.limits().max_collection_items() - result.length() {
raise LimitExceeded(
offset=decoder.offset(),
limit="maximum collection item count",
)
}
let start = decoder.offset()
for _ in 0.. {
let result : Map[String, Datum] = Map([])
let mut count = decoder.read_long()
while count != 0L {
let (entry_count, declared_size) = collection_block(decoder, count)
if entry_count >
decoder.limits().max_collection_items() - result.length() {
raise LimitExceeded(
offset=decoder.offset(),
limit="maximum collection item count",
)
}
let start = decoder.offset()
for _ in 0.. {
let offset = decoder.offset()
let index = decoder.read_int()
if index < 0 || index >= branches.length() {
raise InvalidEncoding(
offset~,
message="union branch index is outside schema",
)
}
Union(index, read_schema(root, branches[index], decoder, depth + 1))
}
@schema.SchemaKind::Fixed(name=_, namespace_=_, aliases=_, size~) =>
Fixed(decoder.read_raw(size))
@schema.SchemaKind::Named(name) =>
match root.resolve_named(name) {
Some(kind) =>
read_schema(root, @schema.Schema::new(kind, name), decoder, depth + 1)
None =>
raise InvalidEncoding(
offset=decoder.offset(),
message="unresolved named schema: \{name}",
)
}
}
}
///|
/// Encode one datum according to an Avro schema.
pub fn encode(schema : @schema.Schema, datum : Datum) -> Bytes raise CodecError {
let encoder = Encoder::new()
write_schema(schema, schema, datum, encoder, 0)
encoder.to_bytes()
}
///|
/// Decode one datum according to an Avro schema and reject trailing bytes.
pub fn decode(
schema : @schema.Schema,
input : Bytes,
limits? : CodecLimits = CodecLimits::new(),
) -> Datum raise CodecError {
let decoder = Decoder::new(input, limits~)
let value = read_schema(schema, schema, decoder, 0)
decoder.ensure_finished()
value
}
///|
/// Encode one datum into a caller-owned stream encoder.
pub fn encode_into(
schema : @schema.Schema,
datum : Datum,
encoder : Encoder,
) -> Unit raise CodecError {
write_schema(schema, schema, datum, encoder, 0)
}
///|
/// Decode one datum from a caller-owned stream decoder, leaving later bytes
/// available for the next datum.
pub fn decode_from(
schema : @schema.Schema,
decoder : Decoder,
) -> Datum raise CodecError {
read_schema(schema, schema, decoder, 0)
}