| 534 | self.reset_read() |
| 535 | |
| 536 | def _deserialize( |
| 537 | self, |
| 538 | buffer: Union[Buffer, bytes], |
| 539 | buffers: Iterable = None, |
| 540 | unsupported_objects: Iterable = None, |
| 541 | ): |
| 542 | if isinstance(buffer, bytes): |
| 543 | buffer = Buffer(buffer) |
| 544 | read_context = self.read_context |
| 545 | reader_index = buffer.get_reader_index() |
| 546 | buffer.set_reader_index(reader_index + 1) |
| 547 | bitmap = buffer.get_int8(reader_index) & 0xFF |
| 548 | if bitmap & ~0b11: |
| 549 | raise ValueError(f"Unsupported root header bitmap 0x{bitmap:02x}") |
| 550 | if bool(bitmap & 1) != self.xlang: |
| 551 | raise ValueError("Header bitmap mismatch at xlang bit") |
| 552 | peer_out_of_band_enabled = bool(bitmap & 2) |
| 553 | if peer_out_of_band_enabled and buffers is None: |
| 554 | raise ValueError("Out-of-band buffers are required by the root header") |
| 555 | if not peer_out_of_band_enabled and buffers is not None: |
| 556 | raise ValueError("Out-of-band buffers were provided for an in-band root payload") |
| 557 | read_context.prepare( |
| 558 | buffer, |
| 559 | buffers=buffers, |
| 560 | unsupported_objects=unsupported_objects, |
| 561 | peer_out_of_band_enabled=peer_out_of_band_enabled, |
| 562 | ) |
| 563 | return read_context.read_ref() |
| 564 | |
| 565 | def reset_write(self): |
| 566 | """ |