(buf: bytes, start: int, end: int)
| 84 | |
| 85 | |
| 86 | def _parse(buf: bytes, start: int, end: int) -> Message: |
| 87 | msg = Message() |
| 88 | pos = start |
| 89 | while pos < end: |
| 90 | tag, pos = _read_varint(buf, pos) |
| 91 | wt = tag & 7 |
| 92 | field_no = tag >> 3 |
| 93 | if wt == WT_VARINT: |
| 94 | val, pos = _read_varint(buf, pos) |
| 95 | msg.fields.append(Field(field_no, wt, val)) |
| 96 | elif wt == WT_I64: |
| 97 | v = buf[pos : pos + 8] |
| 98 | pos += 8 |
| 99 | msg.fields.append(Field(field_no, wt, v)) |
| 100 | elif wt == WT_I32: |
| 101 | v = buf[pos : pos + 4] |
| 102 | pos += 4 |
| 103 | msg.fields.append(Field(field_no, wt, v)) |
| 104 | elif wt == WT_LEN: |
| 105 | length, pos = _read_varint(buf, pos) |
| 106 | data = buf[pos : pos + length] |
| 107 | pos += length |
| 108 | child: Union[bytes, Message] |
| 109 | if _looks_like_msg(data): |
| 110 | try: |
| 111 | child = _parse(data, 0, len(data)) |
| 112 | except Exception: |
| 113 | child = data |
| 114 | else: |
| 115 | child = data |
| 116 | msg.fields.append(Field(field_no, wt, child)) |
| 117 | else: |
| 118 | # SGROUP/EGROUP or unknown: bail out silently with what we have so far |
| 119 | break |
| 120 | return msg |
| 121 | |
| 122 | |
| 123 | def _looks_like_msg(buf: bytes) -> bool: |
no test coverage detected