Coverage for Lib/asyncio/streams.py: 93%
405 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 01:31 +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 res = self._client_connected_cb(reader, writer)
243 if coroutines.iscoroutine(res):
244 def callback(task):
245 if task.cancelled():
246 transport.close()
247 return
248 exc = task.exception()
249 if exc is not None:
250 self._loop.call_exception_handler({
251 'message': 'Unhandled exception in client_connected_cb',
252 'exception': exc,
253 'transport': transport,
254 })
255 transport.close()
257 self._task = self._loop.create_task(res)
258 self._task.add_done_callback(callback)
260 self._strong_reader = None
262 def connection_lost(self, exc):
263 reader = self._stream_reader
264 if reader is not None:
265 if exc is None:
266 reader.feed_eof()
267 else:
268 reader.set_exception(exc)
269 if not self._closed.done():
270 if exc is None:
271 self._closed.set_result(None)
272 else:
273 self._closed.set_exception(exc)
274 super().connection_lost(exc)
275 self._stream_reader_wr = None
276 self._task = None
277 self._transport = None
279 def data_received(self, data):
280 reader = self._stream_reader
281 if reader is not None: 281 ↛ exitline 281 didn't return from function 'data_received' because the condition on line 281 was always true
282 reader.feed_data(data)
284 def eof_received(self):
285 reader = self._stream_reader
286 if reader is not None:
287 reader.feed_eof()
288 if self._over_ssl:
289 # Prevent a warning in SSLProtocol.eof_received:
290 # "returning true from eof_received()
291 # has no effect when using ssl"
292 return False
293 return True
295 def _get_close_waiter(self, stream):
296 return self._closed
298 def __del__(self):
299 # Prevent reports about unhandled exceptions.
300 # Better than self._closed._log_traceback = False hack
301 try:
302 closed = self._closed
303 except AttributeError:
304 pass # failed constructor
305 else:
306 if closed.done() and not closed.cancelled():
307 closed.exception()
310class StreamWriter:
311 """Wraps a Transport.
313 This exposes write(), writelines(), [can_]write_eof(),
314 get_extra_info() and close(). It adds drain() which returns an
315 optional Future on which you can wait for flow control. It also
316 adds a transport property which references the Transport
317 directly.
318 """
320 def __init__(self, transport, protocol, reader, loop):
321 self._transport = transport
322 self._protocol = protocol
323 # drain() expects that the reader has an exception() method
324 assert reader is None or isinstance(reader, StreamReader)
325 self._reader = reader
326 self._loop = loop
328 def __repr__(self):
329 info = [self.__class__.__name__, f'transport={self._transport!r}']
330 if self._reader is not None: 330 ↛ 332line 330 didn't jump to line 332 because the condition on line 330 was always true
331 info.append(f'reader={self._reader!r}')
332 return '<{}>'.format(' '.join(info))
334 @property
335 def transport(self):
336 return self._transport
338 def write(self, data):
339 self._transport.write(data)
341 def writelines(self, data):
342 self._transport.writelines(data)
344 def write_eof(self):
345 return self._transport.write_eof()
347 def can_write_eof(self):
348 return self._transport.can_write_eof()
350 def close(self):
351 return self._transport.close()
353 def is_closing(self):
354 return self._transport.is_closing()
356 async def wait_closed(self):
357 await self._protocol._get_close_waiter(self)
359 def get_extra_info(self, name, default=None):
360 return self._transport.get_extra_info(name, default)
362 async def drain(self):
363 """Flush the write buffer.
365 The intended use is to write
367 w.write(data)
368 await w.drain()
369 """
370 if self._reader is not None:
371 exc = self._reader.exception()
372 if exc is not None: 372 ↛ 373line 372 didn't jump to line 373 because the condition on line 372 was never true
373 raise exc
374 if self._transport.is_closing():
375 # Wait for protocol.connection_lost() call
376 # Raise connection closing error if any,
377 # ConnectionResetError otherwise
378 # Yield to the event loop so connection_lost() may be
379 # called. Without this, _drain_helper() would return
380 # immediately, and code that calls
381 # write(...); await drain()
382 # in a loop would never call connection_lost(), so it
383 # would not see an error when the socket is closed.
384 await sleep(0)
385 await self._protocol._drain_helper()
387 async def start_tls(self, sslcontext, *,
388 server_hostname=None,
389 ssl_handshake_timeout=None,
390 ssl_shutdown_timeout=None):
391 """Upgrade an existing stream-based connection to TLS."""
392 server_side = self._protocol._client_connected_cb is not None
393 protocol = self._protocol
394 await self.drain()
395 new_transport = await self._loop.start_tls( # type: ignore
396 self._transport, protocol, sslcontext,
397 server_side=server_side, server_hostname=server_hostname,
398 ssl_handshake_timeout=ssl_handshake_timeout,
399 ssl_shutdown_timeout=ssl_shutdown_timeout)
400 self._transport = new_transport
401 protocol._replace_transport(new_transport)
403 def __del__(self, warnings=warnings):
404 if not self._transport.is_closing():
405 if self._loop.is_closed(): 405 ↛ 406line 405 didn't jump to line 406 because the condition on line 405 was never true
406 warnings.warn("loop is closed", ResourceWarning)
407 else:
408 self.close()
409 warnings.warn(f"unclosed {self!r}", ResourceWarning)
411class StreamReader:
413 _source_traceback = None
415 def __init__(self, limit=_DEFAULT_LIMIT, loop=None):
416 # The line length limit is a security feature;
417 # it also doubles as half the buffer limit.
419 if limit <= 0:
420 raise ValueError('Limit cannot be <= 0')
422 self._limit = limit
423 if loop is None:
424 self._loop = events.get_event_loop()
425 else:
426 self._loop = loop
427 self._buffer = bytearray()
428 self._eof = False # Whether we're done.
429 self._waiter = None # A future used by _wait_for_data()
430 self._exception = None
431 self._transport = None
432 self._paused = False
433 if self._loop.get_debug():
434 self._source_traceback = format_helpers.extract_stack(
435 sys._getframe(1))
437 def __repr__(self):
438 info = ['StreamReader']
439 if self._buffer:
440 info.append(f'{len(self._buffer)} bytes')
441 if self._eof:
442 info.append('eof')
443 if self._limit != _DEFAULT_LIMIT:
444 info.append(f'limit={self._limit}')
445 if self._waiter:
446 info.append(f'waiter={self._waiter!r}')
447 if self._exception:
448 info.append(f'exception={self._exception!r}')
449 if self._transport:
450 info.append(f'transport={self._transport!r}')
451 if self._paused: 451 ↛ 452line 451 didn't jump to line 452 because the condition on line 451 was never true
452 info.append('paused')
453 return '<{}>'.format(' '.join(info))
455 def exception(self):
456 return self._exception
458 def set_exception(self, exc):
459 self._exception = exc
461 waiter = self._waiter
462 if waiter is not None:
463 self._waiter = None
464 if not waiter.cancelled(): 464 ↛ exitline 464 didn't return from function 'set_exception' because the condition on line 464 was always true
465 waiter.set_exception(exc)
467 def _wakeup_waiter(self):
468 """Wakeup read*() functions waiting for data or EOF."""
469 waiter = self._waiter
470 if waiter is not None:
471 self._waiter = None
472 if not waiter.cancelled(): 472 ↛ exitline 472 didn't return from function '_wakeup_waiter' because the condition on line 472 was always true
473 waiter.set_result(None)
475 def set_transport(self, transport):
476 assert self._transport is None, 'Transport already set'
477 self._transport = transport
479 def _replace_transport(self, transport):
480 assert self._transport is not None, 'Transport not set'
481 self._transport = transport
483 def _maybe_resume_transport(self):
484 if self._paused and len(self._buffer) <= self._limit:
485 self._paused = False
486 self._transport.resume_reading()
488 def feed_eof(self):
489 self._eof = True
490 self._wakeup_waiter()
492 def at_eof(self):
493 """Return True if the buffer is empty and 'feed_eof' was called."""
494 return self._eof and not self._buffer
496 def feed_data(self, data):
497 assert not self._eof, 'feed_data after feed_eof'
499 if not data:
500 return
502 self._buffer.extend(data)
503 self._wakeup_waiter()
505 if (self._transport is not None and
506 not self._paused and
507 len(self._buffer) > 2 * self._limit):
508 try:
509 self._transport.pause_reading()
510 except NotImplementedError:
511 # The transport can't be paused.
512 # We'll just have to buffer all data.
513 # Forget the transport so we don't keep trying.
514 self._transport = None
515 else:
516 self._paused = True
518 async def _wait_for_data(self, func_name):
519 """Wait until feed_data() or feed_eof() is called.
521 If stream was paused, automatically resume it.
522 """
523 # StreamReader uses a future to link the protocol feed_data() method
524 # to a read coroutine. Running two read coroutines at the same time
525 # would have an unexpected behaviour. It would not possible to know
526 # which coroutine would get the next data.
527 if self._waiter is not None: 527 ↛ 528line 527 didn't jump to line 528 because the condition on line 527 was never true
528 raise RuntimeError(
529 f'{func_name}() called while another coroutine is '
530 f'already waiting for incoming data')
532 assert not self._eof, '_wait_for_data after EOF'
534 # Waiting for data while paused will make deadlock, so prevent it.
535 # This is essential for readexactly(n) for case when n > self._limit.
536 if self._paused: 536 ↛ 537line 536 didn't jump to line 537 because the condition on line 536 was never true
537 self._paused = False
538 self._transport.resume_reading()
540 self._waiter = self._loop.create_future()
541 try:
542 await self._waiter
543 finally:
544 self._waiter = None
546 async def readline(self):
547 r"""Read chunk of data from the stream until newline (b'\n') is found.
549 On success, return chunk that ends with newline. If only partial
550 line can be read due to EOF, return incomplete line without
551 terminating newline. When EOF was reached while no bytes read,
552 empty bytes object is returned.
554 If limit is reached, ValueError will be raised. In that case, if
555 newline was found, complete line including newline will be removed
556 from internal buffer. Else, internal buffer will be cleared.
557 Limit is compared against part of the line without newline.
559 If stream was paused, this function will automatically resume it if
560 needed.
561 """
562 sep = b'\n'
563 seplen = len(sep)
564 try:
565 line = await self.readuntil(sep)
566 except exceptions.IncompleteReadError as e:
567 return e.partial
568 except exceptions.LimitOverrunError as e:
569 if self._buffer.startswith(sep, e.consumed):
570 del self._buffer[:e.consumed + seplen]
571 else:
572 self._buffer.clear()
573 self._maybe_resume_transport()
574 raise ValueError(e.args[0])
575 return line
577 async def readuntil(self, separator=b'\n'):
578 """Read data from the stream until ``separator`` is found.
580 On success, the data and separator will be removed from the
581 internal buffer (consumed). Returned data will include the
582 separator at the end.
584 Configured stream limit is used to check result. Limit sets the
585 maximal length of data that can be returned, not counting the
586 separator.
588 If an EOF occurs and the complete separator is still not found,
589 an IncompleteReadError exception will be raised, and the internal
590 buffer will be reset. The IncompleteReadError.partial attribute
591 may contain the separator partially.
593 If the data cannot be read because of over limit, a
594 LimitOverrunError exception will be raised, and the data
595 will be left in the internal buffer, so it can be read again.
597 The ``separator`` may also be a tuple of separators. In this
598 case the return value will be the shortest possible that has any
599 separator as the suffix. For the purposes of LimitOverrunError,
600 the shortest possible separator is considered to be the one that
601 matched.
602 """
603 if isinstance(separator, tuple):
604 # Makes sure shortest matches wins
605 separator = sorted(separator, key=len)
606 else:
607 separator = [separator]
608 if not separator:
609 raise ValueError('Separator should contain at least one element')
610 min_seplen = len(separator[0])
611 max_seplen = len(separator[-1])
612 if min_seplen == 0:
613 raise ValueError('Separator should be at least one-byte string')
615 if self._exception is not None:
616 raise self._exception
618 # Consume whole buffer except last bytes, which length is
619 # one less than max_seplen. Let's check corner cases with
620 # separator[-1]='SEPARATOR':
621 # * we have received almost complete separator (without last
622 # byte). i.e buffer='some textSEPARATO'. In this case we
623 # can safely consume max_seplen - 1 bytes.
624 # * last byte of buffer is first byte of separator, i.e.
625 # buffer='abcdefghijklmnopqrS'. We may safely consume
626 # everything except that last byte, but this require to
627 # analyze bytes of buffer that match partial separator.
628 # This is slow and/or require FSM. For this case our
629 # implementation is not optimal, since require rescanning
630 # of data that is known to not belong to separator. In
631 # real world, separator will not be so long to notice
632 # performance problems. Even when reading MIME-encoded
633 # messages :)
635 # `offset` is the number of bytes from the beginning of the buffer
636 # where there is no occurrence of any `separator`.
637 offset = 0
639 # Loop until we find a `separator` in the buffer, exceed the buffer size,
640 # or an EOF has happened.
641 while True:
642 buflen = len(self._buffer)
644 # Check if we now have enough data in the buffer for shortest
645 # separator to fit.
646 if buflen - offset >= min_seplen:
647 match_start = None
648 match_end = None
649 for sep in separator:
650 isep = self._buffer.find(sep, offset)
652 if isep != -1:
653 # `separator` is in the buffer. `match_start` and
654 # `match_end` will be used later to retrieve the
655 # data.
656 end = isep + len(sep)
657 if match_end is None or end < match_end:
658 match_end = end
659 match_start = isep
660 if match_end is not None:
661 break
663 # see upper comment for explanation.
664 offset = max(0, buflen + 1 - max_seplen)
665 if offset > self._limit:
666 raise exceptions.LimitOverrunError(
667 'Separator is not found, and chunk exceed the limit',
668 offset)
670 # Complete message (with full separator) may be present in buffer
671 # even when EOF flag is set. This may happen when the last chunk
672 # adds data which makes separator be found. That's why we check for
673 # EOF *after* inspecting the buffer.
674 if self._eof:
675 chunk = self._buffer.take_bytes()
676 raise exceptions.IncompleteReadError(chunk, None)
678 # _wait_for_data() will resume reading if stream was paused.
679 await self._wait_for_data('readuntil')
681 if match_start > self._limit:
682 raise exceptions.LimitOverrunError(
683 'Separator is found, but chunk is longer than limit', match_start)
685 chunk = self._buffer.take_bytes(match_end)
686 self._maybe_resume_transport()
687 return chunk
689 async def read(self, n=-1):
690 """Read up to `n` bytes from the stream.
692 If `n` is not provided or set to -1,
693 read until EOF, then return all read bytes.
694 If EOF was received and the internal buffer is empty,
695 return an empty bytes object.
697 If `n` is 0, return an empty bytes object immediately.
699 If `n` is positive, return at most `n` available bytes
700 as soon as at least 1 byte is available in the internal buffer.
701 If EOF is received before any byte is read, return an empty
702 bytes object.
704 Returned value is not limited with limit, configured at stream
705 creation.
707 If stream was paused, this function will automatically resume it if
708 needed.
709 """
711 if self._exception is not None:
712 raise self._exception
714 if n == 0:
715 return b''
717 if n < 0:
718 # This used to just loop creating a new waiter hoping to
719 # collect everything in self._buffer, but that would
720 # deadlock if the subprocess sends more than self.limit
721 # bytes. So just call self.read(self._limit) until EOF.
722 joined = bytearray()
723 while block := await self.read(self._limit):
724 joined += block
725 return joined.take_bytes()
727 if not self._buffer and not self._eof:
728 await self._wait_for_data('read')
730 # This will work right even if buffer is less than n bytes
731 data = self._buffer.take_bytes(min(len(self._buffer), n))
733 self._maybe_resume_transport()
734 return data
736 async def readexactly(self, n):
737 """Read exactly `n` bytes.
739 Raise an IncompleteReadError if EOF is reached before `n` bytes can be
740 read. The IncompleteReadError.partial attribute of the exception will
741 contain the partial read bytes.
743 if n is zero, return empty bytes object.
745 Returned value is not limited with limit, configured at stream
746 creation.
748 If stream was paused, this function will automatically resume it if
749 needed.
750 """
751 if n < 0:
752 raise ValueError('readexactly size can not be less than zero')
754 if self._exception is not None:
755 raise self._exception
757 if n == 0:
758 return b''
760 while len(self._buffer) < n:
761 if self._eof:
762 incomplete = self._buffer.take_bytes()
763 raise exceptions.IncompleteReadError(incomplete, n)
765 await self._wait_for_data('readexactly')
767 data = self._buffer.take_bytes(n)
768 self._maybe_resume_transport()
769 return data
771 def __aiter__(self):
772 return self
774 async def __anext__(self):
775 val = await self.readline()
776 if val == b'':
777 raise StopAsyncIteration
778 return val