Write a WebSocket frame. Args: frame: Frame to write. write: Function that writes bytes. mask: Whether the frame should be masked i.e. whether the write happens on the client side. extensions: List of extensions, appli
(
self,
write: Callable[[bytes], Any],
*,
mask: bool,
extensions: Sequence[extensions.Extension] | None = None,
)
| 120 | ) |
| 121 | |
| 122 | def write( |
| 123 | self, |
| 124 | write: Callable[[bytes], Any], |
| 125 | *, |
| 126 | mask: bool, |
| 127 | extensions: Sequence[extensions.Extension] | None = None, |
| 128 | ) -> None: |
| 129 | """ |
| 130 | Write a WebSocket frame. |
| 131 | |
| 132 | Args: |
| 133 | frame: Frame to write. |
| 134 | write: Function that writes bytes. |
| 135 | mask: Whether the frame should be masked i.e. whether the write |
| 136 | happens on the client side. |
| 137 | extensions: List of extensions, applied in order. |
| 138 | |
| 139 | Raises: |
| 140 | ProtocolError: If the frame contains incorrect values. |
| 141 | |
| 142 | """ |
| 143 | # The frame is written in a single call to write in order to prevent |
| 144 | # TCP fragmentation. See #68 for details. This also makes it safe to |
| 145 | # send frames concurrently from multiple coroutines. |
| 146 | write(self.new_frame.serialize(mask=mask, extensions=extensions)) |
| 147 | |
| 148 | |
| 149 | def prepare_data(data: DataLike) -> tuple[int, BytesLike]: |
no test coverage detected