(self, buffer, length: int)
| 154 | return encoded_meta_string |
| 155 | |
| 156 | def _read_small_meta_string(self, buffer, length: int): |
| 157 | if length == 0: |
| 158 | return EMPTY_ENCODED_META_STRING |
| 159 | encoding = buffer.read_int8() & 0xFF |
| 160 | if encoding > 4: |
| 161 | raise ValueError(f"Unexpected encoding flag: {encoding}") |
| 162 | if length <= 8: |
| 163 | v1 = buffer.read_bytes_as_int64(length) |
| 164 | v2 = 0 |
| 165 | else: |
| 166 | v1 = buffer.read_int64() |
| 167 | v2 = buffer.read_bytes_as_int64(length - 8) |
| 168 | key = (v1, v2, encoding, length) |
| 169 | encoded_meta_string = self._small_encoded_meta_strings.get(key) |
| 170 | if encoded_meta_string is not None: |
| 171 | return encoded_meta_string |
| 172 | hashcode = hash_small_metastring(v1, v2, length, encoding) |
| 173 | reader_index = buffer.get_reader_index() |
| 174 | data = buffer.get_bytes(reader_index - length, length) |
| 175 | encoded_meta_string = self.shared_registry.get_or_create_encoded_meta_string(data, hashcode) |
| 176 | if len(self._small_encoded_meta_strings) < MAX_CACHED_META_STRINGS: |
| 177 | self._small_encoded_meta_strings[key] = encoded_meta_string |
| 178 | return encoded_meta_string |
| 179 | |
| 180 | def _read_big_meta_string(self, buffer, length: int): |
| 181 | hashcode = buffer.read_int64() |
no test coverage detected