Given an iterator that yields raw binary data, iterate over it & yield every event encountered
(self, iterator: Iterator[bytes])
| 425 | self._raw = [] |
| 426 | |
| 427 | def iter_bytes(self, iterator: Iterator[bytes]) -> Iterator[ServerSentEvent]: |
| 428 | """Given an iterator that yields raw binary data, iterate over it & yield every event encountered""" |
| 429 | for chunk in self._iter_chunks(iterator): |
| 430 | # Split before decoding so splitlines() only uses \r and \n |
| 431 | for raw_line in chunk.splitlines(): |
| 432 | line = raw_line.decode("utf-8") |
| 433 | sse = self.decode(line) |
| 434 | if sse: |
| 435 | yield sse |
| 436 | |
| 437 | def _iter_chunks(self, iterator: Iterator[bytes]) -> Iterator[bytes]: |
| 438 | """Given an iterator that yields raw binary data, iterate over it and yield individual SSE chunks""" |
no test coverage detected