Given an iterator that yields raw binary data, iterate over it & yield every event encountered
(self, iterator: Iterator[bytes])
| 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""" |
no test coverage detected