<commit_before>import asyncio
import collections
from typing import List  # noqa
from typing import Awaitable, Callable, Optional, Tuple

from .helpers import set_exception, set_result
from .log import internal_logger
from .typedefs import Byteish


__all__ = (
    'EMPTY_PAYLOAD', 'EofStream', 'StreamReader', 'DataQueue',
    'FlowControlDataQueue')

DEFAULT_LIMIT = 2 ** 16


class EofStream(Exception):
    """eof stream indication."""


class AsyncStreamIterator:

    def __init__(self, read_func: Callable[[], Awaitable[bytes]]) -> None:
        self.read_func = read_func

    def __aiter__(self) -> 'AsyncStreamIterator':
        return self

    async def __anext__(self) -> bytes:
        try:
            rv = await self.read_func()
        except EofStream:
            raise StopAsyncIteration  # NOQA
        if rv == b'':
            raise StopAsyncIteration  # NOQA
        return rv


class ChunkTupleAsyncStreamIterator(AsyncStreamIterator):
    async def __anext__(self) -> bytes:
        rv = await self.read_func()
        if rv == (b'', False):
            raise StopAsyncIteration  # NOQA
        return rv


class AsyncStreamReaderMixin:

    def __aiter__(self) -> AsyncStreamIterator:
        return AsyncStreamIterator(self.readline)  # type: ignore

    def iter_chunked(self, n: int) -> AsyncStreamIterator:
        """Returns an asynchronous iterator that yields chunks of size n.

        Python-3.5 available for Python 3.5+ only
        """
        return AsyncStreamIterator(lambda: self.read(n))  # type: ignore

    def iter_any(self) -> AsyncStreamIterator:
        """Returns an asynchronous iterator that yields all the available
        data as soon as it is received

        Python-3.5 available for Python 3.5+ only
        """
        return AsyncStreamIterator(self.readany)  # type: ignore

    def iter_chunks(self) -> ChunkTupleAsyncStreamIterator:
        """Returns an asynchronous iterator that yields chunks of data
        as they are received by the server. The yielded objects are tuples
        of (bytes, bool) as returned by the StreamReader.readchunk method.

        Python-3.5 available for Python 3.5+ only
        """
        return ChunkTupleAsyncStreamIterator(self.readchunk)  # type: ignore


class StreamReader(AsyncStreamReaderMixin):
    """An enhancement of asyncio.StreamReader.

    Supports asynchronous iteration by line, chunk or as available::
<commit_msg>fix iter_chunks type hint (#3230) (#3232)<commit_after>import asyncio
import collections
from typing import List  # noqa
from typing import Awaitable, Callable, Optional, Tuple

from .helpers import set_exception, set_result
from .log import internal_logger
from .typedefs import Byteish


__all__ = (
    'EMPTY_PAYLOAD', 'EofStream', 'StreamReader', 'DataQueue',
    'FlowControlDataQueue')

DEFAULT_LIMIT = 2 ** 16


class EofStream(Exception):
    """eof stream indication."""


class AsyncStreamIterator:

    def __init__(self, read_func: Callable[[], Awaitable[bytes]]) -> None:
        self.read_func = read_func

    def __aiter__(self) -> 'AsyncStreamIterator':
        return self

    async def __anext__(self) -> bytes:
        try:
            rv = await self.read_func()
        except EofStream:
            raise StopAsyncIteration  # NOQA
        if rv == b'':
            raise StopAsyncIteration  # NOQA
        return rv


class ChunkTupleAsyncStreamIterator:

    def __init__(self, stream: 'StreamReader') -> None:
        self._stream = stream

    def __aiter__(self) -> 'ChunkTupleAsyncStreamIterator':
        return self

    async def __anext__(self) -> Tuple[bytes, bool]:
        rv = await self._stream.readchunk()
        if rv == (b'', False):
            raise StopAsyncIteration  # NOQA
        return rv


class AsyncStreamReaderMixin:

    def __aiter__(self) -> AsyncStreamIterator:
        return AsyncStreamIterator(self.readline)  # type: ignore

    def iter_chunked(self, n: int) -> AsyncStreamIterator:
        """Returns an asynchronous iterator that yields chunks of size n.

        Python-3.5 available for Python 3.5+ only
        """
        return AsyncStreamIterator(lambda: self.read(n))  # type: ignore

    def iter_any(self) -> AsyncStreamIterator:
        """Returns an asynchronous iterator that yields all the available
        data as soon as it is received

        Python-3.5 available for Python 3.5+ only
        """
        return AsyncStreamIterator(self.readany)  # type: ignore

    def iter_chunks(self) -> ChunkTupleAsyncStreamIterator:
        """Returns an asynchronous iterator that yields chunks of data
        as they are received by the server. The yielded objects are tuples
        of (bytes, bool) as returned by the StreamReader.readchunk method.

        Python-3.5 available for Python 3.5+ only
        """
        return ChunkTupleAsyncStreamIterator(self)  # type: ignore


class StreamReader(AsyncStreamReaderMixin):
    """An enhancement of asyncio.StreamReader.

    Supports asynchronous iteration by line, chunk or as available::

{'fraction_non_alphanumeric': 0.08325093929207039, 'fraction_numerical': 0.007514336563179751, 'mean_word_length': 2.9515625, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1411155', 'n_tokens_mistral': 1438, 'n_tokens_neox': 1361, 'n_words': 456}