(self)
| 85 | self.payload = None |
| 86 | |
| 87 | def pprint(self): |
| 88 | s = f"[{self.Names[self.packet_type]}]" |
| 89 | |
| 90 | if self.packet_type == self.CONNECT: |
| 91 | assert self.payload |
| 92 | s += f""" |
| 93 | |
| 94 | Client Id: {self.payload["ClientId"]} |
| 95 | Will Topic: {self.payload.get("WillTopic")} |
| 96 | Will Message: {strutils.bytes_to_escaped_str(self.payload.get("WillMessage", b"None"))} |
| 97 | User Name: {self.payload.get("UserName")} |
| 98 | Password: {strutils.bytes_to_escaped_str(self.payload.get("Password", b"None"))} |
| 99 | """ |
| 100 | elif self.packet_type == self.SUBSCRIBE: |
| 101 | s += " sent topic filters: " |
| 102 | s += ", ".join([f"'{tf}'" for tf in self.topic_filters]) |
| 103 | elif self.packet_type == self.PUBLISH: |
| 104 | assert self.payload |
| 105 | topic_name = strutils.bytes_to_escaped_str(self.topic_name) |
| 106 | payload = strutils.bytes_to_escaped_str(self.payload) |
| 107 | |
| 108 | s += f" '{payload}' to topic '{topic_name}'" |
| 109 | elif self.packet_type in [self.PINGREQ, self.PINGRESP]: |
| 110 | pass |
| 111 | else: |
| 112 | s = f"Packet type {self.Names[self.packet_type]} is not supported yet!" |
| 113 | |
| 114 | return s |
| 115 | |
| 116 | def _parse_length_prefixed_bytes(self, offset): |
| 117 | field_length_bytes = self._packet[offset : offset + 2] |
no test coverage detected