# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# pyformat: disable

"""Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT."""

import re
import json
from dataclasses import dataclass, asdict
from typing import (
    Any,
    Callable,
    Generic,
    List,
    TypeVar,
    Optional,
    Generator,
    AsyncGenerator,
    Tuple,
)
import httpx

T = TypeVar("T")


class Stream(Generic[T]):
    # Holds a reference to the SDK client to avoid it being garbage collected
    # and cause termination of the underlying httpx client.
    client_ref: Optional[object]
    response: httpx.Response
    generator: Generator[T, None, None]
    _closed: bool

    def __init__(
        self,
        response: httpx.Response,
        decoder: Callable[[str], T],
        sentinel: Optional[str] = None,
        client_ref: Optional[object] = None,
        data_required: bool = True,
    ):
        self.response = response
        self.generator = stream_events(
            response, decoder, sentinel, data_required=data_required
        )
        self.client_ref = client_ref
        self._closed = False

    def __iter__(self):
        return self

    def __next__(self):
        if self._closed:
            raise StopIteration
        return next(self.generator)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def close(self):
        self._closed = True
        self.response.close()


class AsyncStream(Generic[T]):
    # Holds a reference to the SDK client to avoid it being garbage collected
    # and cause termination of the underlying httpx client.
    client_ref: Optional[object]
    response: httpx.Response
    generator: AsyncGenerator[T, None]
    _closed: bool

    def __init__(
        self,
        response: httpx.Response,
        decoder: Callable[[str], T],
        sentinel: Optional[str] = None,
        client_ref: Optional[object] = None,
        data_required: bool = True,
    ):
        self.response = response
        self.generator = stream_events_async(
            response, decoder, sentinel, data_required=data_required
        )
        self.client_ref = client_ref
        self._closed = False

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self._closed:
            raise StopAsyncIteration
        return await self.generator.__anext__()

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()

    async def close(self):
        self._closed = True
        await self.response.aclose()


@dataclass
class ServerEvent:
    id: Optional[str] = None
    event: Optional[str] = None
    data: Any = None
    retry: Optional[int] = None


MESSAGE_BOUNDARIES = [
    b"\r\n\r\n",
    b"\r\n\r",
    b"\r\n\n",
    b"\r\r\n",
    b"\n\r\n",
    b"\r\r",
    b"\n\r",
    b"\n\n",
]
MAX_BOUNDARY_LEN = max(len(b) for b in MESSAGE_BOUNDARIES)

UTF8_BOM = b"\xef\xbb\xbf"


async def stream_events_async(
    response: httpx.Response,
    decoder: Callable[[str], T],
    sentinel: Optional[str] = None,
    data_required: bool = True,
) -> AsyncGenerator[T, None]:
    try:
        buffer = bytearray()
        position = 0
        event_id: Optional[str] = None
        async for chunk in response.aiter_bytes():
            if len(buffer) == 0 and chunk.startswith(UTF8_BOM):
                chunk = chunk[len(UTF8_BOM) :]
            old_len = len(buffer)
            buffer += chunk
            search_start = max(position, old_len - MAX_BOUNDARY_LEN + 1)
            for i in range(search_start, len(buffer)):
                char = buffer[i : i + 1]
                seq: Optional[bytes] = None
                if char in [b"\r", b"\n"]:
                    for boundary in MESSAGE_BOUNDARIES:
                        seq = _peek_sequence(i, buffer, boundary)
                        if seq is not None:
                            break
                if seq is None:
                    continue

                block = buffer[position:i]
                position = i + len(seq)
                event, discard, event_id = _parse_event(
                    raw=block,
                    decoder=decoder,
                    sentinel=sentinel,
                    event_id=event_id,
                    data_required=data_required,
                )
                if event is not None:
                    yield event
                if discard:
                    return

            if position > 0:
                buffer = buffer[position:]
                position = 0

        event, discard, _ = _parse_event(
            raw=buffer,
            decoder=decoder,
            sentinel=sentinel,
            event_id=event_id,
            data_required=data_required,
        )
        if event is not None:
            yield event
    finally:
        await response.aclose()


def stream_events(
    response: httpx.Response,
    decoder: Callable[[str], T],
    sentinel: Optional[str] = None,
    data_required: bool = True,
) -> Generator[T, None, None]:
    try:
        buffer = bytearray()
        position = 0
        event_id: Optional[str] = None
        for chunk in response.iter_bytes():
            if len(buffer) == 0 and chunk.startswith(UTF8_BOM):
                chunk = chunk[len(UTF8_BOM) :]
            old_len = len(buffer)
            buffer += chunk
            search_start = max(position, old_len - MAX_BOUNDARY_LEN + 1)
            for i in range(search_start, len(buffer)):
                char = buffer[i : i + 1]
                seq: Optional[bytes] = None
                if char in [b"\r", b"\n"]:
                    for boundary in MESSAGE_BOUNDARIES:
                        seq = _peek_sequence(i, buffer, boundary)
                        if seq is not None:
                            break
                if seq is None:
                    continue

                block = buffer[position:i]
                position = i + len(seq)
                event, discard, event_id = _parse_event(
                    raw=block,
                    decoder=decoder,
                    sentinel=sentinel,
                    event_id=event_id,
                    data_required=data_required,
                )
                if event is not None:
                    yield event
                if discard:
                    return

            if position > 0:
                buffer = buffer[position:]
                position = 0

        event, discard, _ = _parse_event(
            raw=buffer,
            decoder=decoder,
            sentinel=sentinel,
            event_id=event_id,
            data_required=data_required,
        )
        if event is not None:
            yield event
    finally:
        response.close()


def _parse_event(
    *,
    raw: bytearray,
    decoder: Callable[[str], T],
    sentinel: Optional[str] = None,
    event_id: Optional[str] = None,
    data_required: bool = True,
) -> Tuple[Optional[T], bool, Optional[str]]:
    block = raw.decode()
    lines = re.split(r"\r?\n|\r", block)
    publish = False
    event = ServerEvent()
    data_parts: List[str] = []
    for line in lines:
        if not line:
            continue

        delim = line.find(":")
        if delim == 0:
            continue

        field = line
        value = ""
        if delim > 0:
            field = line[0:delim]
            value = line[delim + 1 :] if delim < len(line) - 1 else ""
            if len(value) and value[0] == " ":
                value = value[1:]

        if field == "event":
            event.event = value
            publish = True
        elif field == "data":
            data_parts.append(value)
            publish = True
        elif field == "id":
            publish = True
            if "\x00" not in value:
                event_id = value
        elif field == "retry":
            if value.isdigit():
                event.retry = int(value)
            publish = True

    event.id = event_id
    has_data = bool(data_parts)
    data = "\n".join(data_parts)

    if sentinel and has_data and data == sentinel:
        return None, True, event_id

    # Skip data-less events when data is required
    if not has_data and publish and data_required:
        return None, False, event_id

    if has_data:
        try:
            event.data = json.loads(data)
        except json.JSONDecodeError:
            event.data = data

    out = None
    if publish:
        out_dict = {
            k: v
            for k, v in asdict(event).items()
            if v is not None or (k == "data" and has_data)
        }
        out = decoder(json.dumps(out_dict))

    return out, False, event_id


def _peek_sequence(position: int, buffer: bytearray, sequence: bytes):
    if len(sequence) > (len(buffer) - position):
        return None

    for i, seq in enumerate(sequence):
        if buffer[position + i] != seq:
            return None

    return sequence