| 187 | |
| 188 | |
| 189 | class SSEDecoder: |
| 190 | _data: list[str] |
| 191 | _event: str | None |
| 192 | _retry: int | None |
| 193 | _last_event_id: str | None |
| 194 | |
| 195 | def __init__(self) -> None: |
| 196 | self._event = None |
| 197 | self._data = [] |
| 198 | self._last_event_id = None |
| 199 | self._retry = None |
| 200 | |
| 201 | def iter_bytes(self, iterator: Iterator[bytes]) -> Iterator[ServerSentEvent]: |
| 202 | """Given an iterator that yields raw binary data, iterate over it & yield every event encountered""" |
| 203 | for chunk in self._iter_chunks(iterator): |
| 204 | # Split before decoding so splitlines() only uses \r and \n |
| 205 | for raw_line in chunk.splitlines(): |
| 206 | line = raw_line.decode("utf-8") |
| 207 | sse = self.decode(line) |
| 208 | if sse: |
| 209 | yield sse |
| 210 | |
| 211 | def _iter_chunks(self, iterator: Iterator[bytes]) -> Iterator[bytes]: |
| 212 | """Given an iterator that yields raw binary data, iterate over it and yield individual SSE chunks""" |
| 213 | data = b"" |
| 214 | for chunk in iterator: |
| 215 | for line in chunk.splitlines(keepends=True): |
| 216 | data += line |
| 217 | if data.endswith((b"\r\r", b"\n\n", b"\r\n\r\n")): |
| 218 | yield data |
| 219 | data = b"" |
| 220 | if data: |
| 221 | yield data |
| 222 | |
| 223 | async def aiter_bytes(self, iterator: AsyncIterator[bytes]) -> AsyncIterator[ServerSentEvent]: |
| 224 | """Given an iterator that yields raw binary data, iterate over it & yield every event encountered""" |
| 225 | async for chunk in self._aiter_chunks(iterator): |
| 226 | # Split before decoding so splitlines() only uses \r and \n |
| 227 | for raw_line in chunk.splitlines(): |
| 228 | line = raw_line.decode("utf-8") |
| 229 | sse = self.decode(line) |
| 230 | if sse: |
| 231 | yield sse |
| 232 | |
| 233 | async def _aiter_chunks(self, iterator: AsyncIterator[bytes]) -> AsyncIterator[bytes]: |
| 234 | """Given an iterator that yields raw binary data, iterate over it and yield individual SSE chunks""" |
| 235 | data = b"" |
| 236 | async for chunk in iterator: |
| 237 | for line in chunk.splitlines(keepends=True): |
| 238 | data += line |
| 239 | if data.endswith((b"\r\r", b"\n\n", b"\r\n\r\n")): |
| 240 | yield data |
| 241 | data = b"" |
| 242 | if data: |
| 243 | yield data |
| 244 | |
| 245 | def decode(self, line: str) -> ServerSentEvent | None: |
| 246 | # See: https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation # noqa: E501 |