_memory_streams.py 文件源码

def receive_some(self, max_bytes):
        async with self._receive_conflict_detector:
            # Argument validation
            max_bytes = operator.index(max_bytes)
            if max_bytes < 1:
                raise ValueError("max_bytes must be >= 1")
            # State validation
            if self._receiver_closed:
                raise ClosedStreamError
            # Wake wait_send_all_might_not_block and wait for data
            self._receiver_waiting = True
            self._something_happened()
            try:
                await self._wait_for(lambda: self._data or self._sender_closed)
            finally:
                self._receiver_waiting = False
            # Get data, possibly waking send_all
            if self._data:
                got = self._data[:max_bytes]
                del self._data[:max_bytes]
                self._something_happened()
                return got
            else:
                assert self._sender_closed
                return b""