///|
fn encode(packet : @codec.Packet, limit : Int) -> Bytes raise ClientError {
match @codec.encode_packet(packet) {
Err(e) => raise ProtocolError(e)
Ok(data) => {
if data.length() > limit {
raise ProtocolError("outgoing packet exceeds limit")
}
Bytes::from_array(data)
}
}
}
///|
async fn read_packet(reader : &@io.Reader, limit : Int) -> @codec.Packet {
let header = reader.read_exactly(1)
let data : Array[Byte] = [header[0]]
let mut size = 0
let mut multiplier = 1
for index in 0..<4 {
let digit = reader.read_exactly(1)[0]
data.push(digit)
size += (digit.to_int() & 127) * multiplier
if size + data.length() > limit {
raise ProtocolError("incoming packet exceeds limit")
}
if digit.to_int() < 128 {
if index > 0 && digit == 0 {
raise ProtocolError("noncanonical remaining length")
}
break
}
if index == 3 {
raise ProtocolError("malformed remaining length")
}
multiplier *= 128
}
let body = reader.read_exactly(size)
for byte in body {
data.push(byte)
}
match @codec.decode_packet(data) {
Ok(packet) => packet
Err(reason) => raise ProtocolError(reason)
}
}
///|
fn Config::connect_packet(self : Config) -> @codec.Packet {
@codec.ConnectPacket({
client_id: self.client_id,
username: self.username,
password: self.password,
will_topic: self.will.map(w => w.topic),
will_message: self.will.map(w => w.payload),
keep_alive: self.keep_alive_secs,
flags: {
username: self.username is Some(_),
password: self.password is Some(_),
will_retain: self.will.map(w => w.retain).unwrap_or(false),
will_qos: self.will.map(w => w.qos.wire()).unwrap_or(@codec.QoS0),
will_flag: self.will is Some(_),
clean_session: true,
},
})
}