def pack(self, we_are_a_client = True) -> bytes:
payload: bytes = b''
msg: Union[CtrlMessage, NetMessage, ConnlessMessage]
is_control: bool = False
is_connless: bool = False
for msg in self.messages:
if msg.message_type == 'connless':
is_connless = True
msg = cast(ConnlessMessage, msg)
payload += bytes(msg.message_id)
payload += msg.pack()
elif msg.message_type == 'control':
is_control = True
msg = cast(CtrlMessage, msg)
payload += pack_int(msg.message_id)
payload += msg.pack(we_are_a_client)
else: # game or system message
msg = cast(NetMessage, msg)
msg_payload: bytes = pack_int(
(msg.message_id<<1) |
(int)(msg.system_message)
)
msg_payload += msg.pack()
if msg.header.size is None:
msg.header.size = len(msg_payload)
payload += msg.header.pack()
payload += msg_payload
if self.header.num_chunks is None:
if is_control:
self.header.num_chunks = 0
else:
self.header.num_chunks = len(self.messages)
if is_control:
if self.header.flags.control is None:
self.header.flags.control = True
if is_connless:
if self.header.flags.connless is None:
self.header.flags.connless = True
if self.header.flags.compression:
payload = twnet_parser.huffman.compress(payload)
return self.header.pack() + payload
def pack(self, we_are_a_client = True) -> bytes:
payload: bytes = b''
msg: Union[CtrlMessage, NetMessage, ConnlessMessage]
is_control: bool = False
is_connless: bool = False
for msg in self.messages:
if msg.message_type == 'connless':
is_connless = True
msg = cast(ConnlessMessage, msg)
payload += bytes(msg.message_id)
payload += msg.pack()
elif msg.message_type == 'control':
is_control = True
msg = cast(CtrlMessage, msg)
payload += pack_int(msg.message_id)
payload += msg.pack(we_are_a_client)
else: # game or system message
msg = cast(NetMessage, msg)
msg_payload: bytes = pack_int(
(msg.message_id<<1) |
(int)(msg.system_message)
)
msg_payload += msg.pack()
if msg.header.size is None:
msg.header.size = len(msg_payload)
payload += msg.header.pack()
payload += msg_payload
if self.header.num_chunks is None:
if is_control:
self.header.num_chunks = 0
else:
self.header.num_chunks = len(self.messages)
if is_control:
if self.header.flags.control is None:
self.header.flags.control = True
if is_connless:
if self.header.flags.connless is None:
self.header.flags.connless = True
if self.header.flags.compression:
payload = twnet_parser.huffman.compress(payload)
return self.header.pack() + payload
IsFreeze
GetTile
which is called in IsFreeze
TILE_FREEZE
in the code