a wrapper for createconnection returning a reader writer pair the reader returned is a streamreader instance the writer is a streamwriter instance the arguments are all the usual arguments to createconnection except protocolfactory most common are positional host and port with various optional keyword arguments following additional optional keyword arguments are loop to set the event loop instance to use and limit to set the buffer limit passed to the streamreader if you want to customize the streamreader andor streamreaderprotocol classes just copy the code there s really nothing special here except some convenience start a socket server call back for each client connected the first parameter clientconnectedcb takes two parameters clientreader clientwriter clientreader is a streamreader object while clientwriter is a streamwriter object this parameter can either be a plain callback function or a coroutine if it is a coroutine it will be automatically converted into a task the rest of the arguments are all the usual arguments to loop createserver except protocolfactory most common are positional host and port with various optional keyword arguments following the return value is the same as loop createserver additional optional keyword argument is limit to set the buffer limit passed to the streamreader the return value is the same as loop createserver i e a server object which can be used to stop the service unix domain sockets are supported on this platform similar to openconnection but works with unix domain sockets loop events getrunningloop reader streamreaderlimitlimit looploop protocol streamreaderprotocolreader looploop transport await loop createunixconnection lambda protocol path kwds writer streamwritertransport protocol reader loop return reader writer async def startunixserverclientconnectedcb pathnone limitdefaultlimit kwds reusable flow control logic for streamwriter drain this implements the protocol methods pausewriting resumewriting and connectionlost if the subclass overrides these it must call the super methods streamwriter drain must wait for drainhelper coroutine wake up the writers if currently paused helper class to adapt between protocol and streamreader this is a helper class instead of making streamreader itself a protocol subclass because the streamreader has other potential uses and to prevent the user of the streamreader to accidentally call inappropriate methods of the protocol this is a stream the createserver function keep a strong reference to the reader until a connection is established prevent a warning in sslprotocol eofreceived returning true from eofreceived has no effect when using ssl prevent reports about unhandled exceptions better than self closed logtraceback false hack wraps a transport this exposes write writelines canwriteeof getextrainfo and close it adds drain which returns an optional future on which you can wait for flow control it also adds a transport property which references the transport directly drain expects that the reader has an exception method flush the write buffer the intended use is to write w writedata await w drain wait for protocol connectionlost call raise connection closing error if any connectionreseterror otherwise yield to the event loop so connectionlost may be called without this drainhelper would return immediately and code that calls write await drain in a loop would never call connectionlost so it would not see an error when the socket is closed upgrade an existing streambased connection to tls serverside self protocol clientconnectedcb is not none protocol self protocol await self drain newtransport await self loop starttls type ignore self transport protocol sslcontext serversideserverside serverhostnameserverhostname sslhandshaketimeoutsslhandshaketimeout sslshutdowntimeoutsslshutdowntimeout self transport newtransport protocol replacewriterself def delself warningswarnings if not self transport isclosing if self loop isclosed warnings warnloop is closed resourcewarning else self close warnings warnfunclosed self r resourcewarning class streamreader sourcetraceback none def initself limitdefaultlimit loopnone the line length limit is a security feature it also doubles as half the buffer limit if limit 0 raise valueerror limit cannot be 0 self limit limit if loop is none self loop events geteventloop else self loop loop self buffer bytearray self eof false whether we re done self waiter none a future used by waitfordata self exception none self transport none self paused false if self loop getdebug self sourcetraceback formathelpers extractstack sys getframe1 def reprself info streamreader if self buffer info appendf lenself buffer bytes if self eof info append eof if self limit defaultlimit info appendf limitself limit if self waiter info appendf waiterself waiter r if self exception info appendf exceptionself exception r if self transport info appendf transportself transport r if self paused info append paused return format joininfo def exceptionself return self exception def setexceptionself exc self exception exc waiter self waiter if waiter is not none self waiter none if not waiter cancelled waiter setexceptionexc def wakeupwaiterself return true if the buffer is empty and feedeof was called return self eof and not self buffer def feeddataself data assert not self eof feeddata after feedeof if not data return self buffer extenddata self wakeupwaiter if self transport is not none and not self paused and lenself buffer 2 self limit try self transport pausereading except notimplementederror the transport can t be paused we ll just have to buffer all data forget the transport so we don t keep trying self transport none else self paused true async def waitfordataself funcname streamreader uses a future to link the protocol feeddata method to a read coroutine running two read coroutines at the same time would have an unexpected behaviour it would not possible to know which coroutine would get the next data if self waiter is not none raise runtimeerror f funcname called while another coroutine is f already waiting for incoming data assert not self eof waitfordata after eof waiting for data while paused will make deadlock so prevent it this is essential for readexactlyn for case when n self limit if self paused self paused false self transport resumereading self waiter self loop createfuture try await self waiter finally self waiter none async def readlineself sep b n seplen lensep try line await self readuntilsep except exceptions incompletereaderror as e return e partial except exceptions limitoverrunerror as e if self buffer startswithsep e consumed del self buffer e consumed seplen else self buffer clear self mayberesumetransport raise valueerrore args0 return line async def readuntilself separatorb n seplen lenseparator if seplen 0 raise valueerror separator should be at least onebyte string if self exception is not none raise self exception consume whole buffer except last bytes which length is one less than seplen let s check corner cases with separator separator we have received almost complete separator without last byte i e buffer some textseparato in this case we can safely consume lenseparator 1 bytes last byte of buffer is first byte of separator i e buffer abcdefghijklmnopqrs we may safely consume everything except that last byte but this require to analyze bytes of buffer that match partial separator this is slow andor require fsm for this case our implementation is not optimal since require rescanning of data that is known to not belong to separator in real world separator will not be so long to notice performance problems even when reading mimeencoded messages offset is the number of bytes from the beginning of the buffer where there is no occurrence of separator offset 0 loop until we find separator in the buffer exceed the buffer size or an eof has happened while true buflen lenself buffer check if we now have enough data in the buffer for separator to fit if buflen offset seplen isep self buffer findseparator offset if isep 1 separator is in the buffer isep will be used later to retrieve the data break see upper comment for explanation offset buflen 1 seplen if offset self limit raise exceptions limitoverrunerror separator is not found and chunk exceed the limit offset complete message with full separator may be present in buffer even when eof flag is set this may happen when the last chunk adds data which makes separator be found that s why we check for eof ater inspecting the buffer if self eof chunk bytesself buffer self buffer clear raise exceptions incompletereaderrorchunk none waitfordata will resume reading if stream was paused await self waitfordata readuntil if isep self limit raise exceptions limitoverrunerror separator is found but chunk is longer than limit isep chunk self buffer isep seplen del self buffer isep seplen self mayberesumetransport return byteschunk async def readself n1 if self exception is not none raise self exception if n 0 return b if n 0 this used to just loop creating a new waiter hoping to collect everything in self buffer but that would deadlock if the subprocess sends more than self limit bytes so just call self readself limit until eof blocks while true block await self readself limit if not block break blocks appendblock return b joinblocks if not self buffer and not self eof await self waitfordata read this will work right even if buffer is less than n bytes data bytesmemoryviewself buffer n del self buffer n self mayberesumetransport return data async def readexactlyself n if n 0 raise valueerror readexactly size can not be less than zero if self exception is not none raise self exception if n 0 return b while lenself buffer n if self eof incomplete bytesself buffer self buffer clear raise exceptions incompletereaderrorincomplete n await self waitfordata readexactly if lenself buffer n data bytesself buffer self buffer clear else data bytesmemoryviewself buffer n del self buffer n self mayberesumetransport return data def aiterself return self async def anextself val await self readline if val b raise stopasynciteration return val 64 kib a wrapper for create_connection returning a reader writer pair the reader returned is a streamreader instance the writer is a streamwriter instance the arguments are all the usual arguments to create_connection except protocol_factory most common are positional host and port with various optional keyword arguments following additional optional keyword arguments are loop to set the event loop instance to use and limit to set the buffer limit passed to the streamreader if you want to customize the streamreader and or streamreaderprotocol classes just copy the code there s really nothing special here except some convenience start a socket server call back for each client connected the first parameter client_connected_cb takes two parameters client_reader client_writer client_reader is a streamreader object while client_writer is a streamwriter object this parameter can either be a plain callback function or a coroutine if it is a coroutine it will be automatically converted into a task the rest of the arguments are all the usual arguments to loop create_server except protocol_factory most common are positional host and port with various optional keyword arguments following the return value is the same as loop create_server additional optional keyword argument is limit to set the buffer limit passed to the streamreader the return value is the same as loop create_server i e a server object which can be used to stop the service unix domain sockets are supported on this platform similar to open_connection but works with unix domain sockets similar to start_server but works with unix domain sockets reusable flow control logic for streamwriter drain this implements the protocol methods pause_writing resume_writing and connection_lost if the subclass overrides these it must call the super methods streamwriter drain must wait for _drain_helper coroutine wake up the writer s if currently paused helper class to adapt between protocol and streamreader this is a helper class instead of making streamreader itself a protocol subclass because the streamreader has other potential uses and to prevent the user of the streamreader to accidentally call inappropriate methods of the protocol this is a stream the create_server function keep a strong reference to the reader until a connection is established prevent a warning in sslprotocol eof_received returning true from eof_received has no effect when using ssl prevent reports about unhandled exceptions better than self _closed _log_traceback false hack failed constructor wraps a transport this exposes write writelines can_ write_eof get_extra_info and close it adds drain which returns an optional future on which you can wait for flow control it also adds a transport property which references the transport directly drain expects that the reader has an exception method flush the write buffer the intended use is to write w write data await w drain wait for protocol connection_lost call raise connection closing error if any connectionreseterror otherwise yield to the event loop so connection_lost may be called without this _drain_helper would return immediately and code that calls write await drain in a loop would never call connection_lost so it would not see an error when the socket is closed upgrade an existing stream based connection to tls type ignore the line length limit is a security feature it also doubles as half the buffer limit whether we re done a future used by _wait_for_data wakeup read functions waiting for data or eof return true if the buffer is empty and feed_eof was called the transport can t be paused we ll just have to buffer all data forget the transport so we don t keep trying wait until feed_data or feed_eof is called if stream was paused automatically resume it streamreader uses a future to link the protocol feed_data method to a read coroutine running two read coroutines at the same time would have an unexpected behaviour it would not possible to know which coroutine would get the next data waiting for data while paused will make deadlock so prevent it this is essential for readexactly n for case when n self _limit read chunk of data from the stream until newline b n is found on success return chunk that ends with newline if only partial line can be read due to eof return incomplete line without terminating newline when eof was reached while no bytes read empty bytes object is returned if limit is reached valueerror will be raised in that case if newline was found complete line including newline will be removed from internal buffer else internal buffer will be cleared limit is compared against part of the line without newline if stream was paused this function will automatically resume it if needed read data from the stream until separator is found on success the data and separator will be removed from the internal buffer consumed returned data will include the separator at the end configured stream limit is used to check result limit sets the maximal length of data that can be returned not counting the separator if an eof occurs and the complete separator is still not found an incompletereaderror exception will be raised and the internal buffer will be reset the incompletereaderror partial attribute may contain the separator partially if the data cannot be read because of over limit a limitoverrunerror exception will be raised and the data will be left in the internal buffer so it can be read again consume whole buffer except last bytes which length is one less than seplen let s check corner cases with separator separator we have received almost complete separator without last byte i e buffer some textseparato in this case we can safely consume len separator 1 bytes last byte of buffer is first byte of separator i e buffer abcdefghijklmnopqrs we may safely consume everything except that last byte but this require to analyze bytes of buffer that match partial separator this is slow and or require fsm for this case our implementation is not optimal since require rescanning of data that is known to not belong to separator in real world separator will not be so long to notice performance problems even when reading mime encoded messages offset is the number of bytes from the beginning of the buffer where there is no occurrence of separator loop until we find separator in the buffer exceed the buffer size or an eof has happened check if we now have enough data in the buffer for separator to fit separator is in the buffer isep will be used later to retrieve the data see upper comment for explanation complete message with full separator may be present in buffer even when eof flag is set this may happen when the last chunk adds data which makes separator be found that s why we check for eof ater inspecting the buffer _wait_for_data will resume reading if stream was paused read up to n bytes from the stream if n is not provided or set to 1 read until eof then return all read bytes if eof was received and the internal buffer is empty return an empty bytes object if n is 0 return an empty bytes object immediately if n is positive return at most n available bytes as soon as at least 1 byte is available in the internal buffer if eof is received before any byte is read return an empty bytes object returned value is not limited with limit configured at stream creation if stream was paused this function will automatically resume it if needed this used to just loop creating a new waiter hoping to collect everything in self _buffer but that would deadlock if the subprocess sends more than self limit bytes so just call self read self _limit until eof this will work right even if buffer is less than n bytes read exactly n bytes raise an incompletereaderror if eof is reached before n bytes can be read the incompletereaderror partial attribute of the exception will contain the partial read bytes if n is zero return empty bytes object returned value is not limited with limit configured at stream creation if stream was paused this function will automatically resume it if needed
__all__ = (
    'StreamReader', 'StreamWriter', 'StreamReaderProtocol',
    'open_connection', 'start_server')

import collections
import socket
import sys
import warnings
import weakref

if hasattr(socket, 'AF_UNIX'):
    __all__ += ('open_unix_connection', 'start_unix_server')

from . import coroutines
from . import events
from . import exceptions
from . import format_helpers
from . import protocols
from .log import logger
from .tasks import sleep


_DEFAULT_LIMIT = 2 ** 16  


async def open_connection(host=None, port=None, *,
                          limit=_DEFAULT_LIMIT, **kwds):
    
    loop = events.get_running_loop()
    reader = StreamReader(limit=limit, loop=loop)
    protocol = StreamReaderProtocol(reader, loop=loop)
    transport, _ = await loop.create_connection(
        lambda: protocol, host, port, **kwds)
    writer = StreamWriter(transport, protocol, reader, loop)
    return reader, writer


async def start_server(client_connected_cb, host=None, port=None, *,
                       limit=_DEFAULT_LIMIT, **kwds):
    
    loop = events.get_running_loop()

    def factory():
        reader = StreamReader(limit=limit, loop=loop)
        protocol = StreamReaderProtocol(reader, client_connected_cb,
                                        loop=loop)
        return protocol

    return await loop.create_server(factory, host, port, **kwds)


if hasattr(socket, 'AF_UNIX'):
    

    async def open_unix_connection(path=None, *,
                                   limit=_DEFAULT_LIMIT, **kwds):
        
        loop = events.get_running_loop()

        reader = StreamReader(limit=limit, loop=loop)
        protocol = StreamReaderProtocol(reader, loop=loop)
        transport, _ = await loop.create_unix_connection(
            lambda: protocol, path, **kwds)
        writer = StreamWriter(transport, protocol, reader, loop)
        return reader, writer

    async def start_unix_server(client_connected_cb, path=None, *,
                                limit=_DEFAULT_LIMIT, **kwds):
        
        loop = events.get_running_loop()

        def factory():
            reader = StreamReader(limit=limit, loop=loop)
            protocol = StreamReaderProtocol(reader, client_connected_cb,
                                            loop=loop)
            return protocol

        return await loop.create_unix_server(factory, path, **kwds)


class FlowControlMixin(protocols.Protocol):
    

    def __init__(self, loop=None):
        if loop is None:
            self._loop = events.get_event_loop()
        else:
            self._loop = loop
        self._paused = False
        self._drain_waiters = collections.deque()
        self._connection_lost = False

    def pause_writing(self):
        assert not self._paused
        self._paused = True
        if self._loop.get_debug():
            logger.debug("%r pauses writing", self)

    def resume_writing(self):
        assert self._paused
        self._paused = False
        if self._loop.get_debug():
            logger.debug("%r resumes writing", self)

        for waiter in self._drain_waiters:
            if not waiter.done():
                waiter.set_result(None)

    def connection_lost(self, exc):
        self._connection_lost = True
        
        if not self._paused:
            return

        for waiter in self._drain_waiters:
            if not waiter.done():
                if exc is None:
                    waiter.set_result(None)
                else:
                    waiter.set_exception(exc)

    async def _drain_helper(self):
        if self._connection_lost:
            raise ConnectionResetError('Connection lost')
        if not self._paused:
            return
        waiter = self._loop.create_future()
        self._drain_waiters.append(waiter)
        try:
            await waiter
        finally:
            self._drain_waiters.remove(waiter)

    def _get_close_waiter(self, stream):
        raise NotImplementedError


class StreamReaderProtocol(FlowControlMixin, protocols.Protocol):
    

    _source_traceback = None

    def __init__(self, stream_reader, client_connected_cb=None, loop=None):
        super().__init__(loop=loop)
        if stream_reader is not None:
            self._stream_reader_wr = weakref.ref(stream_reader)
            self._source_traceback = stream_reader._source_traceback
        else:
            self._stream_reader_wr = None
        if client_connected_cb is not None:
            
            
            
            self._strong_reader = stream_reader
        self._reject_connection = False
        self._stream_writer = None
        self._task = None
        self._transport = None
        self._client_connected_cb = client_connected_cb
        self._over_ssl = False
        self._closed = self._loop.create_future()

    @property
    def _stream_reader(self):
        if self._stream_reader_wr is None:
            return None
        return self._stream_reader_wr()

    def _replace_writer(self, writer):
        loop = self._loop
        transport = writer.transport
        self._stream_writer = writer
        self._transport = transport
        self._over_ssl = transport.get_extra_info('sslcontext') is not None

    def connection_made(self, transport):
        if self._reject_connection:
            context = {
                'message': ('An open stream was garbage collected prior to '
                            'establishing network connection; '
                            'call "stream.close()" explicitly.')
            }
            if self._source_traceback:
                context['source_traceback'] = self._source_traceback
            self._loop.call_exception_handler(context)
            transport.abort()
            return
        self._transport = transport
        reader = self._stream_reader
        if reader is not None:
            reader.set_transport(transport)
        self._over_ssl = transport.get_extra_info('sslcontext') is not None
        if self._client_connected_cb is not None:
            self._stream_writer = StreamWriter(transport, self,
                                               reader,
                                               self._loop)
            res = self._client_connected_cb(reader,
                                            self._stream_writer)
            if coroutines.iscoroutine(res):
                def callback(task):
                    exc = task.exception()
                    if exc is not None:
                        self._loop.call_exception_handler({
                            'message': 'Unhandled exception in client_connected_cb',
                            'exception': exc,
                            'transport': transport,
                        })
                        transport.close()

                self._task = self._loop.create_task(res)
                self._task.add_done_callback(callback)

            self._strong_reader = None

    def connection_lost(self, exc):
        reader = self._stream_reader
        if reader is not None:
            if exc is None:
                reader.feed_eof()
            else:
                reader.set_exception(exc)
        if not self._closed.done():
            if exc is None:
                self._closed.set_result(None)
            else:
                self._closed.set_exception(exc)
        super().connection_lost(exc)
        self._stream_reader_wr = None
        self._stream_writer = None
        self._task = None
        self._transport = None

    def data_received(self, data):
        reader = self._stream_reader
        if reader is not None:
            reader.feed_data(data)

    def eof_received(self):
        reader = self._stream_reader
        if reader is not None:
            reader.feed_eof()
        if self._over_ssl:
            
            
            
            return False
        return True

    def _get_close_waiter(self, stream):
        return self._closed

    def __del__(self):
        
        
        try:
            closed = self._closed
        except AttributeError:
            pass  
        else:
            if closed.done() and not closed.cancelled():
                closed.exception()


class StreamWriter:
    

    def __init__(self, transport, protocol, reader, loop):
        self._transport = transport
        self._protocol = protocol
        
        assert reader is None or isinstance(reader, StreamReader)
        self._reader = reader
        self._loop = loop
        self._complete_fut = self._loop.create_future()
        self._complete_fut.set_result(None)

    def __repr__(self):
        info = [self.__class__.__name__, f'transport={self._transport!r}']
        if self._reader is not None:
            info.append(f'reader={self._reader!r}')
        return '<{}>'.format(' '.join(info))

    @property
    def transport(self):
        return self._transport

    def write(self, data):
        self._transport.write(data)

    def writelines(self, data):
        self._transport.writelines(data)

    def write_eof(self):
        return self._transport.write_eof()

    def can_write_eof(self):
        return self._transport.can_write_eof()

    def close(self):
        return self._transport.close()

    def is_closing(self):
        return self._transport.is_closing()

    async def wait_closed(self):
        await self._protocol._get_close_waiter(self)

    def get_extra_info(self, name, default=None):
        return self._transport.get_extra_info(name, default)

    async def drain(self):
        
        if self._reader is not None:
            exc = self._reader.exception()
            if exc is not None:
                raise exc
        if self._transport.is_closing():
            
            
            
            
            
            
            
            
            
            await sleep(0)
        await self._protocol._drain_helper()

    async def start_tls(self, sslcontext, *,
                        server_hostname=None,
                        ssl_handshake_timeout=None,
                        ssl_shutdown_timeout=None):
        
        server_side = self._protocol._client_connected_cb is not None
        protocol = self._protocol
        await self.drain()
        new_transport = await self._loop.start_tls(  
            self._transport, protocol, sslcontext,
            server_side=server_side, server_hostname=server_hostname,
            ssl_handshake_timeout=ssl_handshake_timeout,
            ssl_shutdown_timeout=ssl_shutdown_timeout)
        self._transport = new_transport
        protocol._replace_writer(self)

    def __del__(self, warnings=warnings):
        if not self._transport.is_closing():
            if self._loop.is_closed():
                warnings.warn("loop is closed", ResourceWarning)
            else:
                self.close()
                warnings.warn(f"unclosed {self!r}", ResourceWarning)

class StreamReader:

    _source_traceback = None

    def __init__(self, limit=_DEFAULT_LIMIT, loop=None):
        
        

        if limit <= 0:
            raise ValueError('Limit cannot be <= 0')

        self._limit = limit
        if loop is None:
            self._loop = events.get_event_loop()
        else:
            self._loop = loop
        self._buffer = bytearray()
        self._eof = False    
        self._waiter = None  
        self._exception = None
        self._transport = None
        self._paused = False
        if self._loop.get_debug():
            self._source_traceback = format_helpers.extract_stack(
                sys._getframe(1))

    def __repr__(self):
        info = ['StreamReader']
        if self._buffer:
            info.append(f'{len(self._buffer)} bytes')
        if self._eof:
            info.append('eof')
        if self._limit != _DEFAULT_LIMIT:
            info.append(f'limit={self._limit}')
        if self._waiter:
            info.append(f'waiter={self._waiter!r}')
        if self._exception:
            info.append(f'exception={self._exception!r}')
        if self._transport:
            info.append(f'transport={self._transport!r}')
        if self._paused:
            info.append('paused')
        return '<{}>'.format(' '.join(info))

    def exception(self):
        return self._exception

    def set_exception(self, exc):
        self._exception = exc

        waiter = self._waiter
        if waiter is not None:
            self._waiter = None
            if not waiter.cancelled():
                waiter.set_exception(exc)

    def _wakeup_waiter(self):
        
        waiter = self._waiter
        if waiter is not None:
            self._waiter = None
            if not waiter.cancelled():
                waiter.set_result(None)

    def set_transport(self, transport):
        assert self._transport is None, 'Transport already set'
        self._transport = transport

    def _maybe_resume_transport(self):
        if self._paused and len(self._buffer) <= self._limit:
            self._paused = False
            self._transport.resume_reading()

    def feed_eof(self):
        self._eof = True
        self._wakeup_waiter()

    def at_eof(self):
        
        return self._eof and not self._buffer

    def feed_data(self, data):
        assert not self._eof, 'feed_data after feed_eof'

        if not data:
            return

        self._buffer.extend(data)
        self._wakeup_waiter()

        if (self._transport is not None and
                not self._paused and
                len(self._buffer) > 2 * self._limit):
            try:
                self._transport.pause_reading()
            except NotImplementedError:
                
                
                
                self._transport = None
            else:
                self._paused = True

    async def _wait_for_data(self, func_name):
        
        
        
        
        
        if self._waiter is not None:
            raise RuntimeError(
                f'{func_name}() called while another coroutine is '
                f'already waiting for incoming data')

        assert not self._eof, '_wait_for_data after EOF'

        
        
        if self._paused:
            self._paused = False
            self._transport.resume_reading()

        self._waiter = self._loop.create_future()
        try:
            await self._waiter
        finally:
            self._waiter = None

    async def readline(self):
        
        sep = b'\n'
        seplen = len(sep)
        try:
            line = await self.readuntil(sep)
        except exceptions.IncompleteReadError as e:
            return e.partial
        except exceptions.LimitOverrunError as e:
            if self._buffer.startswith(sep, e.consumed):
                del self._buffer[:e.consumed + seplen]
            else:
                self._buffer.clear()
            self._maybe_resume_transport()
            raise ValueError(e.args[0])
        return line

    async def readuntil(self, separator=b'\n'):
        
        seplen = len(separator)
        if seplen == 0:
            raise ValueError('Separator should be at least one-byte string')

        if self._exception is not None:
            raise self._exception

        
        
        
        
        
        
        
        
        
        
        
        
        
        
        
        

        
        
        offset = 0

        
        
        while True:
            buflen = len(self._buffer)

            
            
            if buflen - offset >= seplen:
                isep = self._buffer.find(separator, offset)

                if isep != -1:
                    
                    
                    break

                
                offset = buflen + 1 - seplen
                if offset > self._limit:
                    raise exceptions.LimitOverrunError(
                        'Separator is not found, and chunk exceed the limit',
                        offset)

            
            
            
            
            if self._eof:
                chunk = bytes(self._buffer)
                self._buffer.clear()
                raise exceptions.IncompleteReadError(chunk, None)

            
            await self._wait_for_data('readuntil')

        if isep > self._limit:
            raise exceptions.LimitOverrunError(
                'Separator is found, but chunk is longer than limit', isep)

        chunk = self._buffer[:isep + seplen]
        del self._buffer[:isep + seplen]
        self._maybe_resume_transport()
        return bytes(chunk)

    async def read(self, n=-1):
        

        if self._exception is not None:
            raise self._exception

        if n == 0:
            return b''

        if n < 0:
            
            
            
            
            blocks = []
            while True:
                block = await self.read(self._limit)
                if not block:
                    break
                blocks.append(block)
            return b''.join(blocks)

        if not self._buffer and not self._eof:
            await self._wait_for_data('read')

        
        data = bytes(memoryview(self._buffer)[:n])
        del self._buffer[:n]

        self._maybe_resume_transport()
        return data

    async def readexactly(self, n):
        
        if n < 0:
            raise ValueError('readexactly size can not be less than zero')

        if self._exception is not None:
            raise self._exception

        if n == 0:
            return b''

        while len(self._buffer) < n:
            if self._eof:
                incomplete = bytes(self._buffer)
                self._buffer.clear()
                raise exceptions.IncompleteReadError(incomplete, n)

            await self._wait_for_data('readexactly')

        if len(self._buffer) == n:
            data = bytes(self._buffer)
            self._buffer.clear()
        else:
            data = bytes(memoryview(self._buffer)[:n])
            del self._buffer[:n]
        self._maybe_resume_transport()
        return data

    def __aiter__(self):
        return self

    async def __anext__(self):
        val = await self.readline()
        if val == b'':
            raise StopAsyncIteration
        return val