WebSocket frame. Attributes: opcode: Opcode. data: Payload data. fin: FIN bit. rsv1: RSV1 bit. rsv2: RSV2 bit. rsv3: RSV3 bit. Only these fields are needed. The MASK bit, payload length and masking-key are handled on the fly when par
| 121 | |
| 122 | @dataclasses.dataclass |
| 123 | class Frame: |
| 124 | """ |
| 125 | WebSocket frame. |
| 126 | |
| 127 | Attributes: |
| 128 | opcode: Opcode. |
| 129 | data: Payload data. |
| 130 | fin: FIN bit. |
| 131 | rsv1: RSV1 bit. |
| 132 | rsv2: RSV2 bit. |
| 133 | rsv3: RSV3 bit. |
| 134 | |
| 135 | Only these fields are needed. The MASK bit, payload length and masking-key |
| 136 | are handled on the fly when parsing and serializing frames. |
| 137 | |
| 138 | """ |
| 139 | |
| 140 | opcode: Opcode |
| 141 | data: BytesLike |
| 142 | fin: bool = True |
| 143 | rsv1: bool = False |
| 144 | rsv2: bool = False |
| 145 | rsv3: bool = False |
| 146 | |
| 147 | # Configure if you want to see more in logs. Should be a multiple of 3. |
| 148 | MAX_LOG_SIZE = int(os.environ.get("WEBSOCKETS_MAX_LOG_SIZE", "75")) |
| 149 | |
| 150 | def __str__(self) -> str: |
| 151 | """ |
| 152 | Return a human-readable representation of a frame. |
| 153 | |
| 154 | """ |
| 155 | coding = None |
| 156 | length = f"{len(self.data)} byte{'' if len(self.data) == 1 else 's'}" |
| 157 | non_final = "" if self.fin else "continued" |
| 158 | |
| 159 | if self.opcode is OP_TEXT: |
| 160 | # Decoding only the beginning and the end is needlessly hard. |
| 161 | # Decode the entire payload then elide later if necessary. |
| 162 | data = repr(bytes(self.data).decode()) |
| 163 | elif self.opcode is OP_BINARY: |
| 164 | # We'll show at most the first 16 bytes and the last 8 bytes. |
| 165 | # Encode just what we need, plus two dummy bytes to elide later. |
| 166 | binary = self.data |
| 167 | if len(binary) > self.MAX_LOG_SIZE // 3: |
| 168 | cut = (self.MAX_LOG_SIZE // 3 - 1) // 3 # by default cut = 8 |
| 169 | binary = b"".join([binary[: 2 * cut], b"\x00\x00", binary[-cut:]]) |
| 170 | data = " ".join(f"{byte:02x}" for byte in binary) |
| 171 | elif self.opcode is OP_CLOSE: |
| 172 | data = str(Close.parse(self.data)) |
| 173 | elif self.data: |
| 174 | # We don't know if a Continuation frame contains text or binary. |
| 175 | # Ping and Pong frames could contain UTF-8. |
| 176 | # Attempt to decode as UTF-8 and display it as text; fallback to |
| 177 | # binary. If self.data is a memoryview, it has no decode() method, |
| 178 | # which raises AttributeError. |
| 179 | try: |
| 180 | data = repr(bytes(self.data).decode()) |
searching dependent graphs…