Parser

__init__(on_parse, max_event_size, dispatch_empty_named_events) -> None

def __init__(
    self,
    on_parse: Callable[[AnyEvent], object],
    max_event_size: int,
    dispatch_empty_named_events: bool,
) -> None:
    self._on_parse = on_parse
    self._max_event_size = max_event_size
    self._dispatch_empty_named_events = dispatch_empty_named_events
    # Bumped by ``reset``, so ``feed`` stops when the callback resets.
    self._generation = 0
    self.reset()

reset() -> None

def reset(self) -> None:
    """Drop the pending line and event and the last event ID; the next
    chunk may start with a byte-order mark again."""
    self._generation += 1
    self._first_chunk = True
    self._pending: list[str] = []  # pieces of the unfinished line
    self._pending_size = 0
    self._discard_lf = False  # the last line ended with ``\r``
    self._last_event_id: str | None = None
    self._event_name: str | None = None
    self._data: list[str] = []
    self._data_size = 0

feed(chunk) -> SseError | None

def feed(self, chunk: str) -> SseError | None:
    """Parse ``chunk``. Returns an ``SseError`` (and resets) when the
    pending event passes ``max_event_size``. If the callback resets the
    parser, the rest of the chunk is dropped; if it raises, the
    exception propagates and the rest of the chunk is dropped."""
    if self._first_chunk and chunk.startswith(_BOM):
        chunk = chunk[len(_BOM) :]
    self._first_chunk = False
    generation = self._generation
    length = len(chunk)
    position = 0
    while position < length:
        if self._discard_lf:
            self._discard_lf = False
            if chunk[position] == "\n":
                position += 1
                continue
        found = _LINE_BREAK.search(chunk, position)
        if found is None:
            self._pending.append(chunk[position:])
            self._pending_size += length - position
            break
        end = found.start()
        self._discard_lf = chunk[end] == "\r"
        line = chunk[position:end]
        if self._pending:
            self._pending.append(line)
            line = "".join(self._pending)
            self._pending = []
            self._pending_size = 0
        self._parse_line(line)
        if self._generation != generation:
            return None
        position = end + 1
    if self._pending_size + self._data_size > self._max_event_size:
        error = SseError(EventTooLarge(self._max_event_size))
        self.reset()
        return error
    return None