(self, event: events.TextMessage | events.BytesMessage)
| 40 | self.max_length = max_length |
| 41 | |
| 42 | def extend(self, event: events.TextMessage | events.BytesMessage) -> None: |
| 43 | if self.value is None: |
| 44 | self.value = StringIO() if isinstance(event, events.TextMessage) else BytesIO() |
| 45 | self.value.write(event.data) class="cm"># type: ignore[arg-type] |
| 46 | class="cm"># `ws_max_size` is a byte budget, so count UTF-8 bytes for text. |
| 47 | self.length += len(event.data.encode()) if isinstance(event, events.TextMessage) else len(event.data) |
| 48 | if self.length > self.max_length: |
| 49 | raise FrameTooLargeError |
| 50 | |
| 51 | def clear(self) -> None: |
| 52 | self.value = None |
no test coverage detected