(
message, opcode, use_mask=False, noheader=False, is_fin=True, ZLibBackend=None
)
| 47 | |
| 48 | |
| 49 | def build_frame( |
| 50 | message, opcode, use_mask=False, noheader=False, is_fin=True, ZLibBackend=None |
| 51 | ): |
| 52 | # Send a frame over the websocket with message as its payload. |
| 53 | compress = False |
| 54 | if ZLibBackend: |
| 55 | compress = True |
| 56 | compressobj = ZLibBackend.compressobj(wbits=-9) |
| 57 | message = compressobj.compress(message) |
| 58 | message = message + compressobj.flush(ZLibBackend.Z_SYNC_FLUSH) |
| 59 | if message.endswith(WS_DEFLATE_TRAILING): |
| 60 | message = message[:-4] |
| 61 | msg_length = len(message) |
| 62 | if use_mask: # pragma: no cover |
| 63 | mask_bit = 0x80 |
| 64 | else: |
| 65 | mask_bit = 0 |
| 66 | |
| 67 | if is_fin: |
| 68 | header_first_byte = 0x80 | opcode |
| 69 | else: |
| 70 | header_first_byte = opcode |
| 71 | |
| 72 | if compress: |
| 73 | header_first_byte |= 0x40 |
| 74 | |
| 75 | if msg_length < 126: |
| 76 | header = PACK_LEN1(header_first_byte, msg_length | mask_bit) |
| 77 | elif msg_length < (1 << 16): # pragma: no cover |
| 78 | header = PACK_LEN2(header_first_byte, 126 | mask_bit, msg_length) |
| 79 | else: |
| 80 | header = PACK_LEN3(header_first_byte, 127 | mask_bit, msg_length) |
| 81 | |
| 82 | if use_mask: # pragma: no cover |
| 83 | mask = random.randrange(0, 0xFFFFFFFF) |
| 84 | mask = mask.to_bytes(4, "big") |
| 85 | message = bytearray(message) |
| 86 | websocket_mask(mask, message) |
| 87 | if noheader: |
| 88 | return message |
| 89 | else: |
| 90 | return header + mask + message |
| 91 | else: |
| 92 | if noheader: |
| 93 | return message |
| 94 | else: |
| 95 | return header + message |
| 96 | |
| 97 | |
| 98 | def build_close_frame(code=1000, message=b"", noheader=False): |
no test coverage detected