Given an iterator that yields raw binary data, iterate over it & yield every event encountered
(self, iterator: Iterator[bytes])
| 381 | self._raw = [] |
| 382 | |
| 383 | def iter_bytes(self, iterator: Iterator[bytes]) -> Iterator[ServerSentEvent]: |
| 384 | """Given an iterator that yields raw binary data, iterate over it & yield every event encountered""" |
| 385 | for chunk in self._iter_chunks(iterator): |
| 386 | # Split before decoding so splitlines() only uses \r and \n |
| 387 | for raw_line in chunk.splitlines(): |
| 388 | line = raw_line.decode("utf-8") |
| 389 | sse = self.decode(line) |
| 390 | if sse: |
| 391 | yield sse |
| 392 | |
| 393 | def _iter_chunks(self, iterator: Iterator[bytes]) -> Iterator[bytes]: |
| 394 | """Given an iterator that yields raw binary data, iterate over it and yield individual SSE chunks""" |
no test coverage detected