///|
/// RPC message envelope. message_type: 1=call, 2=reply, 3=exception, 4=oneway.
pub(all) struct Message {
name : String
message_type : Int
sequence_id : Int
body : Value
} derive(Debug, Eq)
///|
/// Binary and Compact v1 envelopes. Binary writes strict headers by default.
pub fn encode_message(
message : Message,
protocol : Protocol,
strict_write? : Bool = true,
) -> Bytes raise CodecError {
if message.message_type < 1 || message.message_type > 4 {
raise Invalid("invalid RPC message type")
}
if message.body.kind() != StructKind {
raise Invalid("RPC body must be a struct")
}
let name = @utf8.encode(message.name)
if name.length() > 1024 {
raise Invalid("RPC method name exceeds 1024 bytes")
}
let compact = protocol == CompactProtocol
let out : Array[Byte] = []
if compact {
out.push(130)
out.push(((message.message_type << 5) | 1).to_byte())
varint(out, message.sequence_id.reinterpret_as_uint().to_uint64())
write_value(out, Binary(name), true, 0)
} else if strict_write {
fixed(out, 2147549184UL | message.message_type.to_uint64(), 4, false)
write_value(out, Binary(name), false, 0)
fixed(out, message.sequence_id.reinterpret_as_uint().to_uint64(), 4, false)
} else {
write_value(out, Binary(name), false, 0)
out.push(message.message_type.to_byte())
fixed(out, message.sequence_id.reinterpret_as_uint().to_uint64(), 4, false)
}
write_value(out, message.body, compact, 0)
if out.length() > 1048576 {
raise Invalid("RPC message exceeds one MiB")
}
Bytes::from_array(out)
}
///|
pub fn decode_message(
data : Bytes,
protocol : Protocol,
strict_read? : Bool = true,
) -> Message raise CodecError {
if data.length() > 1048576 {
raise Invalid("RPC message exceeds one MiB")
}
let compact = protocol == CompactProtocol
let reader : Reader = { data, pos: 0, nodes: 0, compact, }
let (message_type, sequence_id, name_data) = if compact {
if reader.byte() != 130 {
raise Invalid("invalid Compact protocol id")
}
let version_type = reader.byte()
if (version_type & 31) != 1 {
raise Invalid("unsupported Compact version")
}
let message_type = version_type >> 5
let seq = reader.varint()
if seq > 4294967295UL {
raise Invalid("sequence id overflow")
}
let sequence_id = seq.to_uint().reinterpret_as_int()
let name_data = match reader.value(BinaryKind, 0) {
Binary(bytes) => bytes
_ => raise Invalid("invalid method name")
}
(message_type, sequence_id, name_data)
} else {
let version_type = reader.fixed(4, false)
if (version_type & 2147483648UL) != 0UL {
if (version_type & 4294901760UL) != 2147549184UL {
raise Invalid("unsupported Binary version")
}
let message_type = (version_type & 255UL).to_int()
let name_data = match reader.value(BinaryKind, 0) {
Binary(bytes) => bytes
_ => raise Invalid("invalid method name")
}
let sequence_id = reader.fixed(4, false).to_uint().reinterpret_as_int()
(message_type, sequence_id, name_data)
} else {
if strict_read {
raise Invalid("expected strict Binary v1 header")
}
if version_type > 1024UL ||
version_type > (data.length() - reader.pos).to_uint64() {
raise Invalid("invalid legacy method length")
}
let end = reader.pos + version_type.to_int()
let name_data = data[reader.pos:end].to_owned()
reader.pos = end
let message_type = reader.byte()
let sequence_id = reader.fixed(4, false).to_uint().reinterpret_as_int()
(message_type, sequence_id, name_data)
}
}
if message_type < 1 || message_type > 4 {
raise Invalid("invalid RPC message type")
}
if name_data.length() > 1024 {
raise Invalid("RPC method name exceeds 1024 bytes")
}
let name = @utf8.decode(name_data) catch {
_ => raise Invalid("invalid UTF-8 method name")
}
let body = reader.value(StructKind, 0)
if reader.pos != data.length() {
raise Invalid("trailing RPC bytes")
}
{ name, message_type, sequence_id, body, }
}