(
self, fin: bool, opcode: int, data: bytes, flags: int = 0
)
| 1028 | ) |
| 1029 | |
| 1030 | def _write_frame( |
| 1031 | self, fin: bool, opcode: int, data: bytes, flags: int = 0 |
| 1032 | ) -> "Future[None]": |
| 1033 | data_len = len(data) |
| 1034 | if opcode & 0x8: |
| 1035 | # All control frames MUST have a payload length of 125 |
| 1036 | # bytes or less and MUST NOT be fragmented. |
| 1037 | if not fin: |
| 1038 | raise ValueError("control frames may not be fragmented") |
| 1039 | if data_len > 125: |
| 1040 | raise ValueError("control frame payloads may not exceed 125 bytes") |
| 1041 | if fin: |
| 1042 | finbit = self.FIN |
| 1043 | else: |
| 1044 | finbit = 0 |
| 1045 | frame = struct.pack("B", finbit | opcode | flags) |
| 1046 | if self.mask_outgoing: |
| 1047 | mask_bit = 0x80 |
| 1048 | else: |
| 1049 | mask_bit = 0 |
| 1050 | if data_len < 126: |
| 1051 | frame += struct.pack("B", data_len | mask_bit) |
| 1052 | elif data_len <= 0xFFFF: |
| 1053 | frame += struct.pack("!BH", 126 | mask_bit, data_len) |
| 1054 | else: |
| 1055 | frame += struct.pack("!BQ", 127 | mask_bit, data_len) |
| 1056 | if self.mask_outgoing: |
| 1057 | mask = os.urandom(4) |
| 1058 | data = mask + _websocket_mask(mask, data) |
| 1059 | frame += data |
| 1060 | self._wire_bytes_out += len(frame) |
| 1061 | return self.stream.write(frame) |
| 1062 | |
| 1063 | def write_message( |
| 1064 | self, message: Union[str, bytes, Dict[str, Any]], binary: bool = False |
no test coverage detected