Coverage for Lib/asyncio/streams.py: 93%
411 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-25 03:29 +0000
1__all__ = (
2 'StreamReader', 'StreamWriter', 'StreamReaderProtocol',
3 'open_connection', 'start_server')
5import collections
6import socket
7import sys
8import warnings
9import weakref
11if hasattr(socket, 'AF_UNIX'): 11 ↛ 14line 11 didn't jump to line 14 because the condition on line 11 was always true
12 __all__ += ('open_unix_connection', 'start_unix_server')
14from . import coroutines
15from . import events
16from . import exceptions
17from . import format_helpers
18from . import protocols
19from .log import logger
20from .tasks import sleep
23_DEFAULT_LIMIT = 2 ** 16 # 64 KiB
26async def open_connection(host=None, port=None, *,
27 limit=_DEFAULT_LIMIT, **kwds):
28 """A wrapper for create_connection() returning a (reader, writer) pair.
30 The reader returned is a StreamReader instance; the writer is a
31 StreamWriter instance.
33 The arguments are all the usual arguments to create_connection()
34 except protocol_factory; most common are positional host and port,
35 with various optional keyword arguments following.
37 Additional optional keyword arguments are loop (to set the event loop
38 instance to use) and limit (to set the buffer limit passed to the
39 StreamReader).
41 (If you want to customize the StreamReader and/or
42 StreamReaderProtocol classes, just copy the code -- there's
43 really nothing special here except some convenience.)
44 """
45 loop = events.get_running_loop()
46 reader = StreamReader(limit=limit, loop=loop)
47 protocol = StreamReaderProtocol(reader, loop=loop)
48 transport, _ = await loop.create_connection(
49 lambda: protocol, host, port, **kwds)
50 writer = StreamWriter(transport, protocol, reader, loop)
51 return reader, writer
54async def start_server(client_connected_cb, host=None, port=None, *,
55 limit=_DEFAULT_LIMIT, **kwds):
56 """Start a socket server, call back for each client connected.
58 The first parameter, `client_connected_cb`, takes two parameters:
59 client_reader, client_writer. client_reader is a StreamReader
60 object, while client_writer is a StreamWriter object. This
61 parameter can either be a plain callback function or a coroutine;
62 if it is a coroutine, it will be automatically converted into a
63 Task.
65 The rest of the arguments are all the usual arguments to
66 loop.create_server() except protocol_factory; most common are
67 positional host and port, with various optional keyword arguments
68 following. The return value is the same as loop.create_server().
70 Additional optional keyword argument is limit (to set the buffer
71 limit passed to the StreamReader).
73 The return value is the same as loop.create_server(), i.e. a
74 Server object which can be used to stop the service.
75 """
76 loop = events.get_running_loop()
78 def factory():
79 reader = StreamReader(limit=limit, loop=loop)
80 protocol = StreamReaderProtocol(reader, client_connected_cb,
81 loop=loop)
82 return protocol
84 return await loop.create_server(factory, host, port, **kwds)
87if hasattr(socket, 'AF_UNIX'): 87 ↛ 116line 87 didn't jump to line 116 because the condition on line 87 was always true
88 # UNIX Domain Sockets are supported on this platform
90 async def open_unix_connection(path=None, *,
91 limit=_DEFAULT_LIMIT, **kwds):
92 """Similar to `open_connection` but works with UNIX Domain Sockets."""
93 loop = events.get_running_loop()
95 reader = StreamReader(limit=limit, loop=loop)
96 protocol = StreamReaderProtocol(reader, loop=loop)
97 transport, _ = await loop.create_unix_connection(
98 lambda: protocol, path, **kwds)
99 writer = StreamWriter(transport, protocol, reader, loop)
100 return reader, writer
102 async def start_unix_server(client_connected_cb, path=None, *,
103 limit=_DEFAULT_LIMIT, **kwds):
104 """Similar to `start_server` but works with UNIX Domain Sockets."""
105 loop = events.get_running_loop()
107 def factory():
108 reader = StreamReader(limit=limit, loop=loop)
109 protocol = StreamReaderProtocol(reader, client_connected_cb,
110 loop=loop)
111 return protocol
113 return await loop.create_unix_server(factory, path, **kwds)
116class FlowControlMixin(protocols.Protocol):
117 """Reusable flow control logic for StreamWriter.drain().
119 This implements the protocol methods pause_writing(),
120 resume_writing() and connection_lost(). If the subclass overrides
121 these it must call the super methods.
123 StreamWriter.drain() must wait for _drain_helper() coroutine.
124 """
126 def __init__(self, loop=None):
127 if loop is None:
128 self._loop = events.get_event_loop()
129 else:
130 self._loop = loop
131 self._paused = False
132 self._drain_waiters = collections.deque()
133 self._connection_lost = False
135 def pause_writing(self):
136 assert not self._paused
137 self._paused = True
138 if self._loop.get_debug():
139 logger.debug("%r pauses writing", self)
141 def resume_writing(self):
142 assert self._paused
143 self._paused = False
144 if self._loop.get_debug(): 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true
145 logger.debug("%r resumes writing", self)
147 for waiter in self._drain_waiters:
148 if not waiter.done(): 148 ↛ 147line 148 didn't jump to line 147 because the condition on line 148 was always true
149 waiter.set_result(None)
151 def connection_lost(self, exc):
152 self._connection_lost = True
153 # Wake up the writer(s) if currently paused.
154 if not self._paused:
155 return
157 for waiter in self._drain_waiters:
158 if not waiter.done():
159 if exc is None: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true
160 waiter.set_result(None)
161 else:
162 waiter.set_exception(exc)
164 async def _drain_helper(self):
165 if self._connection_lost:
166 raise ConnectionResetError('Connection lost')
167 if not self._paused:
168 return
169 waiter = self._loop.create_future()
170 self._drain_waiters.append(waiter)
171 try:
172 await waiter
173 finally:
174 self._drain_waiters.remove(waiter)
176 def _get_close_waiter(self, stream):
177 raise NotImplementedError
180class StreamReaderProtocol(FlowControlMixin, protocols.Protocol):
181 """Helper class to adapt between Protocol and StreamReader.
183 (This is a helper class instead of making StreamReader itself a
184 Protocol subclass, because the StreamReader has other potential
185 uses, and to prevent the user of the StreamReader to accidentally
186 call inappropriate methods of the protocol.)
187 """
189 _source_traceback = None
191 def __init__(self, stream_reader, client_connected_cb=None, loop=None):
192 super().__init__(loop=loop)
193 if stream_reader is not None: 193 ↛ 197line 193 didn't jump to line 197 because the condition on line 193 was always true
194 self._stream_reader_wr = weakref.ref(stream_reader)
195 self._source_traceback = stream_reader._source_traceback
196 else:
197 self._stream_reader_wr = None
198 if client_connected_cb is not None:
199 # This is a stream created by the `create_server()` function.
200 # Keep a strong reference to the reader until a connection
201 # is established.
202 self._strong_reader = stream_reader
203 self._reject_connection = False
204 self._task = None
205 self._transport = None
206 self._client_connected_cb = client_connected_cb
207 self._over_ssl = False
208 self._closed = self._loop.create_future()
210 @property
211 def _stream_reader(self):
212 if self._stream_reader_wr is None:
213 return None
214 return self._stream_reader_wr()
216 def _replace_transport(self, transport):
217 self._transport = transport
218 self._over_ssl = transport.get_extra_info('sslcontext') is not None
219 reader = self._stream_reader
220 if reader is not None: 220 ↛ exitline 220 didn't return from function '_replace_transport' because the condition on line 220 was always true
221 reader._replace_transport(transport)
223 def connection_made(self, transport):
224 if self._reject_connection: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true
225 context = {
226 'message': ('An open stream was garbage collected prior to '
227 'establishing network connection; '
228 'call "stream.close()" explicitly.')
229 }
230 if self._source_traceback:
231 context['source_traceback'] = self._source_traceback
232 self._loop.call_exception_handler(context)
233 transport.abort()
234 return
235 self._transport = transport
236 reader = self._stream_reader
237 if reader is not None: 237 ↛ 239line 237 didn't jump to line 239 because the condition on line 237 was always true
238 reader.set_transport(transport)
239 self._over_ssl = transport.get_extra_info('sslcontext') is not None
240 if self._client_connected_cb is not None:
241 writer = StreamWriter(transport, self, reader, self._loop)
242 try:
243 res = self._client_connected_cb(reader, writer)
244 except Exception as exc:
245 self._loop.call_exception_handler({
246 'message': 'Unhandled exception in client_connected_cb',
247 'exception': exc,
248 'transport': transport,
249 })
250 transport.close()
251 self._strong_reader = None
252 return
253 if coroutines.iscoroutine(res):
254 def callback(task):
255 if task.cancelled():
256 transport.close()
257 return
258 exc = task.exception()
259 if exc is not None:
260 self._loop.call_exception_handler({
261 'message': 'Unhandled exception in client_connected_cb',
262 'exception': exc,
263 'transport': transport,
264 })
265 transport.close()
267 self._task = self._loop.create_task(res)
268 self._task.add_done_callback(callback)
270 self._strong_reader = None
272 def connection_lost(self, exc):
273 reader = self._stream_reader
274 if reader is not None:
275 if exc is None:
276 reader.feed_eof()
277 else:
278 reader.set_exception(exc)
279 if not self._closed.done():
280 if exc is None:
281 self._closed.set_result(None)
282 else:
283 self._closed.set_exception(exc)
284 super().connection_lost(exc)
285 self._stream_reader_wr = None
286 self._task = None
287 self._transport = None
289 def data_received(self, data):
290 reader = self._stream_reader
291 if reader is not None: 291 ↛ exitline 291 didn't return from function 'data_received' because the condition on line 291 was always true
292 reader.feed_data(data)
294 def eof_received(self):
295 reader = self._stream_reader
296 if reader is not None:
297 reader.feed_eof()
298 if self._over_ssl:
299 # Prevent a warning in SSLProtocol.eof_received:
300 # "returning true from eof_received()
301 # has no effect when using ssl"
302 return False
303 return True
305 def _get_close_waiter(self, stream):
306 return self._closed
308 def __del__(self):
309 # Prevent reports about unhandled exceptions.
310 # Better than self._closed._log_traceback = False hack
311 try:
312 closed = self._closed
313 except AttributeError:
314 pass # failed constructor
315 else:
316 if closed.done() and not closed.cancelled():
317 closed.exception()
320class StreamWriter:
321 """Wraps a Transport.
323 This exposes write(), writelines(), [can_]write_eof(),
324 get_extra_info() and close(). It adds drain() which returns an
325 optional Future on which you can wait for flow control. It also
326 adds a transport property which references the Transport
327 directly.
328 """
330 def __init__(self, transport, protocol, reader, loop):
331 self._transport = transport
332 self._protocol = protocol
333 # drain() expects that the reader has an exception() method
334 assert reader is None or isinstance(reader, StreamReader)
335 self._reader = reader
336 self._loop = loop
338 def __repr__(self):
339 info = [self.__class__.__name__, f'transport={self._transport!r}']
340 if self._reader is not None: 340 ↛ 342line 340 didn't jump to line 342 because the condition on line 340 was always true
341 info.append(f'reader={self._reader!r}')
342 return '<{}>'.format(' '.join(info))
344 @property
345 def transport(self):
346 return self._transport
348 def write(self, data):
349 self._transport.write(data)
351 def writelines(self, data):
352 self._transport.writelines(data)
354 def write_eof(self):
355 return self._transport.write_eof()
357 def can_write_eof(self):
358 return self._transport.can_write_eof()
360 def close(self):
361 return self._transport.close()
363 def is_closing(self):
364 return self._transport.is_closing()
366 async def wait_closed(self):
367 await self._protocol._get_close_waiter(self)
369 def get_extra_info(self, name, default=None):
370 return self._transport.get_extra_info(name, default)
372 async def drain(self):
373 """Flush the write buffer.
375 The intended use is to write
377 w.write(data)
378 await w.drain()
379 """
380 if self._reader is not None:
381 exc = self._reader.exception()
382 if exc is not None: 382 ↛ 383line 382 didn't jump to line 383 because the condition on line 382 was never true
383 raise exc
384 if self._transport.is_closing():
385 # Wait for protocol.connection_lost() call
386 # Raise connection closing error if any,
387 # ConnectionResetError otherwise
388 # Yield to the event loop so connection_lost() may be
389 # called. Without this, _drain_helper() would return
390 # immediately, and code that calls
391 # write(...); await drain()
392 # in a loop would never call connection_lost(), so it
393 # would not see an error when the socket is closed.
394 await sleep(0)
395 await self._protocol._drain_helper()
397 async def start_tls(self, sslcontext, *,
398 server_hostname=None,
399 ssl_handshake_timeout=None,
400 ssl_shutdown_timeout=None):
401 """Upgrade an existing stream-based connection to TLS."""
402 server_side = self._protocol._client_connected_cb is not None
403 protocol = self._protocol
404 await self.drain()
405 new_transport = await self._loop.start_tls( # type: ignore
406 self._transport, protocol, sslcontext,
407 server_side=server_side, server_hostname=server_hostname,
408 ssl_handshake_timeout=ssl_handshake_timeout,
409 ssl_shutdown_timeout=ssl_shutdown_timeout)
410 self._transport = new_transport
411 protocol._replace_transport(new_transport)
413 def __del__(self, warnings=warnings):
414 if not self._transport.is_closing():
415 if self._loop.is_closed(): 415 ↛ 416line 415 didn't jump to line 416 because the condition on line 415 was never true
416 warnings.warn("loop is closed", ResourceWarning)
417 else:
418 self.close()
419 warnings.warn(f"unclosed {self!r}", ResourceWarning)
421class StreamReader:
423 _source_traceback = None
425 def __init__(self, limit=_DEFAULT_LIMIT, loop=None):
426 # The line length limit is a security feature;
427 # it also doubles as half the buffer limit.
429 if limit <= 0:
430 raise ValueError('Limit cannot be <= 0')
432 self._limit = limit
433 if loop is None:
434 self._loop = events.get_event_loop()
435 else:
436 self._loop = loop
437 self._buffer = bytearray()
438 self._eof = False # Whether we're done.
439 self._waiter = None # A future used by _wait_for_data()
440 self._exception = None
441 self._transport = None
442 self._paused = False
443 if self._loop.get_debug():
444 self._source_traceback = format_helpers.extract_stack(
445 sys._getframe(1))
447 def __repr__(self):
448 info = ['StreamReader']
449 if self._buffer:
450 info.append(f'{len(self._buffer)} bytes')
451 if self._eof:
452 info.append('eof')
453 if self._limit != _DEFAULT_LIMIT:
454 info.append(f'limit={self._limit}')
455 if self._waiter:
456 info.append(f'waiter={self._waiter!r}')
457 if self._exception:
458 info.append(f'exception={self._exception!r}')
459 if self._transport:
460 info.append(f'transport={self._transport!r}')
461 if self._paused: 461 ↛ 462line 461 didn't jump to line 462 because the condition on line 461 was never true
462 info.append('paused')
463 return '<{}>'.format(' '.join(info))
465 def exception(self):
466 return self._exception
468 def set_exception(self, exc):
469 self._exception = exc
471 waiter = self._waiter
472 if waiter is not None:
473 self._waiter = None
474 if not waiter.cancelled(): 474 ↛ exitline 474 didn't return from function 'set_exception' because the condition on line 474 was always true
475 waiter.set_exception(exc)
477 def _wakeup_waiter(self):
478 """Wakeup read*() functions waiting for data or EOF."""
479 waiter = self._waiter
480 if waiter is not None:
481 self._waiter = None
482 if not waiter.cancelled(): 482 ↛ exitline 482 didn't return from function '_wakeup_waiter' because the condition on line 482 was always true
483 waiter.set_result(None)
485 def set_transport(self, transport):
486 assert self._transport is None, 'Transport already set'
487 self._transport = transport
489 def _replace_transport(self, transport):
490 assert self._transport is not None, 'Transport not set'
491 self._transport = transport
493 def _maybe_resume_transport(self):
494 if self._paused and len(self._buffer) <= self._limit:
495 self._paused = False
496 self._transport.resume_reading()
498 def feed_eof(self):
499 self._eof = True
500 self._wakeup_waiter()
502 def at_eof(self):
503 """Return True if the buffer is empty and 'feed_eof' was called."""
504 return self._eof and not self._buffer
506 def feed_data(self, data):
507 assert not self._eof, 'feed_data after feed_eof'
509 if not data:
510 return
512 self._buffer.extend(data)
513 self._wakeup_waiter()
515 if (self._transport is not None and
516 not self._paused and
517 len(self._buffer) > 2 * self._limit):
518 try:
519 self._transport.pause_reading()
520 except NotImplementedError:
521 # The transport can't be paused.
522 # We'll just have to buffer all data.
523 # Forget the transport so we don't keep trying.
524 self._transport = None
525 else:
526 self._paused = True
528 async def _wait_for_data(self, func_name):
529 """Wait until feed_data() or feed_eof() is called.
531 If stream was paused, automatically resume it.
532 """
533 # StreamReader uses a future to link the protocol feed_data() method
534 # to a read coroutine. Running two read coroutines at the same time
535 # would have an unexpected behaviour. It would not possible to know
536 # which coroutine would get the next data.
537 if self._waiter is not None: 537 ↛ 538line 537 didn't jump to line 538 because the condition on line 537 was never true
538 raise RuntimeError(
539 f'{func_name}() called while another coroutine is '
540 f'already waiting for incoming data')
542 assert not self._eof, '_wait_for_data after EOF'
544 # Waiting for data while paused will make deadlock, so prevent it.
545 # This is essential for readexactly(n) for case when n > self._limit.
546 if self._paused: 546 ↛ 547line 546 didn't jump to line 547 because the condition on line 546 was never true
547 self._paused = False
548 self._transport.resume_reading()
550 self._waiter = self._loop.create_future()
551 try:
552 await self._waiter
553 finally:
554 self._waiter = None
556 async def readline(self):
557 r"""Read chunk of data from the stream until newline (b'\n') is found.
559 On success, return chunk that ends with newline. If only partial
560 line can be read due to EOF, return incomplete line without
561 terminating newline. When EOF was reached while no bytes read,
562 empty bytes object is returned.
564 If limit is reached, ValueError will be raised. In that case, if
565 newline was found, complete line including newline will be removed
566 from internal buffer. Else, internal buffer will be cleared.
567 Limit is compared against part of the line without newline.
569 If stream was paused, this function will automatically resume it if
570 needed.
571 """
572 sep = b'\n'
573 seplen = len(sep)
574 try:
575 line = await self.readuntil(sep)
576 except exceptions.IncompleteReadError as e:
577 return e.partial
578 except exceptions.LimitOverrunError as e:
579 if self._buffer.startswith(sep, e.consumed):
580 del self._buffer[:e.consumed + seplen]
581 else:
582 self._buffer.clear()
583 self._maybe_resume_transport()
584 raise ValueError(e.args[0])
585 return line
587 async def readuntil(self, separator=b'\n'):
588 """Read data from the stream until ``separator`` is found.
590 On success, the data and separator will be removed from the
591 internal buffer (consumed). Returned data will include the
592 separator at the end.
594 Configured stream limit is used to check result. Limit sets the
595 maximal length of data that can be returned, not counting the
596 separator.
598 If an EOF occurs and the complete separator is still not found,
599 an IncompleteReadError exception will be raised, and the internal
600 buffer will be reset. The IncompleteReadError.partial attribute
601 may contain the separator partially.
603 If the data cannot be read because of over limit, a
604 LimitOverrunError exception will be raised, and the data
605 will be left in the internal buffer, so it can be read again.
607 The ``separator`` may also be a tuple of separators. In this
608 case the return value will be the shortest possible that has any
609 separator as the suffix. For the purposes of LimitOverrunError,
610 the shortest possible separator is considered to be the one that
611 matched.
612 """
613 if isinstance(separator, tuple):
614 # Makes sure shortest matches wins
615 separator = sorted(separator, key=len)
616 else:
617 separator = [separator]
618 if not separator:
619 raise ValueError('Separator should contain at least one element')
620 min_seplen = len(separator[0])
621 max_seplen = len(separator[-1])
622 if min_seplen == 0:
623 raise ValueError('Separator should be at least one-byte string')
625 if self._exception is not None:
626 raise self._exception
628 # Consume whole buffer except last bytes, which length is
629 # one less than max_seplen. Let's check corner cases with
630 # separator[-1]='SEPARATOR':
631 # * we have received almost complete separator (without last
632 # byte). i.e buffer='some textSEPARATO'. In this case we
633 # can safely consume max_seplen - 1 bytes.
634 # * last byte of buffer is first byte of separator, i.e.
635 # buffer='abcdefghijklmnopqrS'. We may safely consume
636 # everything except that last byte, but this require to
637 # analyze bytes of buffer that match partial separator.
638 # This is slow and/or require FSM. For this case our
639 # implementation is not optimal, since require rescanning
640 # of data that is known to not belong to separator. In
641 # real world, separator will not be so long to notice
642 # performance problems. Even when reading MIME-encoded
643 # messages :)
645 # `offset` is the number of bytes from the beginning of the buffer
646 # where there is no occurrence of any `separator`.
647 offset = 0
649 # Loop until we find a `separator` in the buffer, exceed the buffer size,
650 # or an EOF has happened.
651 while True:
652 buflen = len(self._buffer)
654 # Check if we now have enough data in the buffer for shortest
655 # separator to fit.
656 if buflen - offset >= min_seplen:
657 match_start = None
658 match_end = None
659 for sep in separator:
660 isep = self._buffer.find(sep, offset)
662 if isep != -1:
663 # `separator` is in the buffer. `match_start` and
664 # `match_end` will be used later to retrieve the
665 # data.
666 end = isep + len(sep)
667 if match_end is None or end < match_end:
668 match_end = end
669 match_start = isep
670 if match_end is not None:
671 break
673 # see upper comment for explanation.
674 offset = max(0, buflen + 1 - max_seplen)
675 if offset > self._limit:
676 raise exceptions.LimitOverrunError(
677 'Separator is not found, and chunk exceed the limit',
678 offset)
680 # Complete message (with full separator) may be present in buffer
681 # even when EOF flag is set. This may happen when the last chunk
682 # adds data which makes separator be found. That's why we check for
683 # EOF *after* inspecting the buffer.
684 if self._eof:
685 chunk = self._buffer.take_bytes()
686 raise exceptions.IncompleteReadError(chunk, None)
688 # _wait_for_data() will resume reading if stream was paused.
689 await self._wait_for_data('readuntil')
691 if match_start > self._limit:
692 raise exceptions.LimitOverrunError(
693 'Separator is found, but chunk is longer than limit', match_start)
695 chunk = self._buffer.take_bytes(match_end)
696 self._maybe_resume_transport()
697 return chunk
699 async def read(self, n=-1):
700 """Read up to `n` bytes from the stream.
702 If `n` is not provided or set to -1,
703 read until EOF, then return all read bytes.
704 If EOF was received and the internal buffer is empty,
705 return an empty bytes object.
707 If `n` is 0, return an empty bytes object immediately.
709 If `n` is positive, return at most `n` available bytes
710 as soon as at least 1 byte is available in the internal buffer.
711 If EOF is received before any byte is read, return an empty
712 bytes object.
714 Returned value is not limited with limit, configured at stream
715 creation.
717 If stream was paused, this function will automatically resume it if
718 needed.
719 """
721 if self._exception is not None:
722 raise self._exception
724 if n == 0:
725 return b''
727 if n < 0:
728 # This used to just loop creating a new waiter hoping to
729 # collect everything in self._buffer, but that would
730 # deadlock if the subprocess sends more than self.limit
731 # bytes. So just call self.read(self._limit) until EOF.
732 joined = bytearray()
733 while block := await self.read(self._limit):
734 joined += block
735 return joined.take_bytes()
737 if not self._buffer and not self._eof:
738 await self._wait_for_data('read')
740 # This will work right even if buffer is less than n bytes
741 data = self._buffer.take_bytes(min(len(self._buffer), n))
743 self._maybe_resume_transport()
744 return data
746 async def readexactly(self, n):
747 """Read exactly `n` bytes.
749 Raise an IncompleteReadError if EOF is reached before `n` bytes can be
750 read. The IncompleteReadError.partial attribute of the exception will
751 contain the partial read bytes.
753 if n is zero, return empty bytes object.
755 Returned value is not limited with limit, configured at stream
756 creation.
758 If stream was paused, this function will automatically resume it if
759 needed.
760 """
761 if n < 0:
762 raise ValueError('readexactly size can not be less than zero')
764 if self._exception is not None:
765 raise self._exception
767 if n == 0:
768 return b''
770 while len(self._buffer) < n:
771 if self._eof:
772 incomplete = self._buffer.take_bytes()
773 raise exceptions.IncompleteReadError(incomplete, n)
775 await self._wait_for_data('readexactly')
777 data = self._buffer.take_bytes(n)
778 self._maybe_resume_transport()
779 return data
781 def __aiter__(self):
782 return self
784 async def __anext__(self):
785 val = await self.readline()
786 if val == b'':
787 raise StopAsyncIteration
788 return val