<commit_before>import asyncio

from asyncio import coroutine

DEFAULT_LIMIT = 2**16

@coroutine
def open_connection(host, port, loop = None):
    if loop is None:
        loop = asyncio.get_event_loop()

    transport, protocol = yield from loop.create_connection(
        lambda : StreamingProtocol(loop = loop),
        host, port
    )

    return (protocol.reader, protocol.writer)

@coroutine
def start_server(connection_callback, host = None, port = None, loop = None):
    if loop is None:
        loop = asyncio.get_event_loop()

    server = yield from loop.create_server(
        lambda: StreamingProtocol(connection_callback, loop = loop),
        host, port
    )

    return server

class StreamReader:
    def __init__(self, transport, limit = DEFAULT_LIMIT, loop = None):
        assert transport

        self._transport = transport
        self._loop = loop or asyncio.get_event_loop() 

        self._buffer = bytearray()
        self._eof = False
        self._pending = None

        self._limit = limit
        self._paused = False

        self._exception = None

    @coroutine
    def _wait(self, parameter, timeout = None):
        if self._pending is not None and not self._pending.done():
            raise RuntimeError("another read call already pending")

        event = asyncio.Future(loop = self._loop)
        self._pending = (parameter, event)

        try:
            yield from asyncio.wait_for(event, timeout, loop = self._loop)
        finally:
            self._pending = None

    def _maybe_pause(self):
        if not self._paused and len(self._buffer) > self._limit:
            try:
                self._transport.pause_reading()
            except NotImplementedError:
                raise RuntimeError("transport cannot be paused")

            self._paused = True

    def _maybe_resume(self):
        if self._paused and len(self._buffer) < self._limit:
            try:
                self._transport.resume_reading()
            except NotImplementedError:
                raise RuntimeError("transport cannot be resumed")

            self._paused = False

    def _feed(self, chunk):
        assert isinstance(chunk, bytes)
        assert chunk

        assert not self._eof

        self._buffer.extend(chunk)
        
        # test pending read calls
        if self._pending:
            parameter, event = self._pending<commit_msg>fix: pausing with read_until call pending now raise an error<commit_after>import asyncio

from asyncio import coroutine

DEFAULT_LIMIT = 2**16

@coroutine
def open_connection(host, port, loop = None):
    if loop is None:
        loop = asyncio.get_event_loop()

    transport, protocol = yield from loop.create_connection(
        lambda : StreamingProtocol(loop = loop),
        host, port
    )

    return (protocol.reader, protocol.writer)

@coroutine
def start_server(connection_callback, host = None, port = None, loop = None):
    if loop is None:
        loop = asyncio.get_event_loop()

    server = yield from loop.create_server(
        lambda: StreamingProtocol(connection_callback, loop = loop),
        host, port
    )

    return server

class StreamReader:
    def __init__(self, transport, limit = DEFAULT_LIMIT, loop = None):
        assert transport

        self._transport = transport
        self._loop = loop or asyncio.get_event_loop() 

        self._buffer = bytearray()
        self._eof = False
        self._pending = None

        self._limit = limit
        self._paused = False

        self._exception = None

    @coroutine
    def _wait(self, parameter, timeout = None):
        if self._pending is not None and not self._pending.done():
            raise RuntimeError("another read call already pending")

        event = asyncio.Future(loop = self._loop)
        self._pending = (parameter, event)

        try:
            yield from asyncio.wait_for(event, timeout, loop = self._loop)
        finally:
            self._pending = None

    def _maybe_pause(self):
        if not self._paused and len(self._buffer) > self._limit:
            try:
                self._transport.pause_reading()
            except NotImplementedError:
                raise RuntimeError("transport cannot be paused")

            # pausing with read_until call pending 
            if self._pending and isinstance(self._pending[0], bytes):
                error = BufferError("buffer limit reached during read_until call")
                self._pending[1].set_exception(error)

            self._paused = True

    def _maybe_resume(self):
        if self._paused and len(self._buffer) < self._limit:
            try:
                self._transport.resume_reading()
            except NotImplementedError:
                raise RuntimeError("transport cannot be resumed")

            self._paused = False

    def _feed(self, chunk):
        assert isinstance(chunk, bytes)
        assert chunk

        assert not self._eof

        self._buffer.extend(chunk)
        
        # test pending read calls
        if self._pending:
            parameter, event = self._pending
{'fraction_non_alphanumeric': 0.07915930072677274, 'fraction_numerical': 0.001571400510705166, 'mean_word_length': 2.174563591022444, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '2229277', 'n_tokens_mistral': 1378, 'n_tokens_neox': 1323, 'n_words': 396}