我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用asyncio.StreamReader()。
def stream_from_fd(fd, loop): """Recieve a streamer for a given file descriptor.""" reader = asyncio.StreamReader(loop=loop) protocol = asyncio.StreamReaderProtocol(reader, loop=loop) waiter = asyncio.futures.Future(loop=loop) transport = UnixFileDescriptorTransport( loop=loop, fileno=fd, protocol=protocol, waiter=waiter, ) try: yield from waiter except: transport.close() raise if loop.get_debug(): logger.debug("Read fd %r connected: (%r, %r)", fd, transport, protocol) return reader, transport
def test_model_base_error_handling_with_patch_and_with_nested_delete(self, patch_method, post_method, stream, session): stream.feed_data(b'[{}]') stream.feed_eof() request = SwaggerRequest('/model1/1/', 'patch', body=stream, headers={'content-type': 'application/json'}) await post_method(request, session) stream = asyncio.StreamReader(loop=session.loop) body = {'model2_': {'id': 1, '_operation': 'delete'}} stream.feed_data(ujson.dumps(body).encode()) stream.feed_eof() request = SwaggerRequest('/model1/1/', 'patch', path_params={'id': 1}, body=stream, headers={'content-type': 'application/json'}) resp = await patch_method(request, session) assert resp.status_code == 400 assert ujson.loads(resp.body) == { 'instance': [body], 'message': "Can't execute nested 'delete' operation" }
def test_model_base_error_handling_with_patch_and_with_nested_remove(self, patch_method, post_method, stream, session): stream.feed_data(b'[{}]') stream.feed_eof() request = SwaggerRequest('/model1/1/', 'patch', body=stream, headers={'content-type': 'application/json'}) await post_method(request, session) stream = asyncio.StreamReader(loop=session.loop) body = {'model2_': {'id': 1, '_operation': 'remove'}} stream.feed_data(ujson.dumps(body).encode()) stream.feed_eof() request = SwaggerRequest('/model1/1/', 'patch', path_params={'id': 1}, body=stream, headers={'content-type': 'application/json'}) resp = await patch_method(request, session) assert resp.status_code == 400 assert ujson.loads(resp.body) == { 'instance': [body], 'message': "Can't execute nested 'remove' operation" }
def test_model_base_error_handling_with_patch_and_with_nested_update(self, patch_method, post_method, stream, session): stream.feed_data(b'[{}]') stream.feed_eof() request = SwaggerRequest('/model1/1/', 'patch', body=stream, headers={'content-type': 'application/json'}) await post_method(request, session) stream = asyncio.StreamReader(loop=session.loop) body = {'model2_': {'id': 1, '_operation': 'update'}} stream.feed_data(ujson.dumps(body).encode()) stream.feed_eof() request = SwaggerRequest('/model1/1/', 'patch', path_params={'id': 1}, body=stream, headers={'content-type': 'application/json'}) resp = await patch_method(request, session) assert resp.status_code == 400 assert ujson.loads(resp.body) == { 'instance': [body], 'message': "Can't execute nested 'update' operation" }
def test_readexactly(self): # Read exact number of bytes. stream = asyncio.StreamReader(loop=self.loop) n = 2 * len(self.DATA) read_task = asyncio.Task(stream.readexactly(n), loop=self.loop) def cb(): stream.feed_data(self.DATA) stream.feed_data(self.DATA) stream.feed_data(self.DATA) self.loop.call_soon(cb) data = self.loop.run_until_complete(read_task) self.assertEqual(self.DATA + self.DATA, data) self.assertEqual(self.DATA, stream._buffer)
def test_readexactly_eof(self): # Read exact number of bytes (eof). stream = asyncio.StreamReader(loop=self.loop) n = 2 * len(self.DATA) read_task = asyncio.Task(stream.readexactly(n), loop=self.loop) def cb(): stream.feed_data(self.DATA) stream.feed_eof() self.loop.call_soon(cb) with self.assertRaises(asyncio.IncompleteReadError) as cm: self.loop.run_until_complete(read_task) self.assertEqual(cm.exception.partial, self.DATA) self.assertEqual(cm.exception.expected, n) self.assertEqual(str(cm.exception), '18 bytes read on a total of 36 expected bytes') self.assertEqual(b'', stream._buffer)
def open_pipe_connection( path=None, *, loop=None, limit=DEFAULT_LIMIT, **kwargs ): """ Connect to a server using a Windows named pipe. """ path = path.replace('/', '\\') loop = loop or asyncio.get_event_loop() reader = asyncio.StreamReader(limit=limit, loop=loop) protocol = asyncio.StreamReaderProtocol(reader, loop=loop) transport, _ = await loop.create_pipe_connection( lambda: protocol, path, **kwargs ) writer = asyncio.StreamWriter(transport, protocol, reader, loop) return reader, writer
def connection_made(self, transport): """ A peer is now connected and we receive an instance of the underlying :class:`asyncio.Transport`. We :class:`asyncio.StreamReader` is created and the transport is associated before the initial HTTP handshake is undertaken. """ #self.transport = transport #self.stream = asyncio.StreamReader() #self.stream.set_transport(transport) asyncio.StreamReaderProtocol.connection_made(self, transport) # Let make it concurrent for others to tag along f = asyncio.async(self.handle_initial_handshake()) f.add_done_callback(self.terminated)
def test_readuntil_multi_chunks(self): stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(b'lineAAA') data = self.loop.run_until_complete(stream.readuntil(separator=b'AAA')) self.assertEqual(b'lineAAA', data) self.assertEqual(b'', stream._buffer) stream.feed_data(b'lineAAA') data = self.loop.run_until_complete(stream.readuntil(b'AAA')) self.assertEqual(b'lineAAA', data) self.assertEqual(b'', stream._buffer) stream.feed_data(b'lineAAAxxx') data = self.loop.run_until_complete(stream.readuntil(b'AAA')) self.assertEqual(b'lineAAA', data) self.assertEqual(b'xxx', stream._buffer)
def test_readuntil_limit_found_sep(self): stream = asyncio.StreamReader(loop=self.loop, limit=3) stream.feed_data(b'some dataAA') with self.assertRaisesRegex(asyncio.LimitOverrunError, 'not found') as cm: self.loop.run_until_complete(stream.readuntil(b'AAA')) self.assertEqual(b'some dataAA', stream._buffer) stream.feed_data(b'A') with self.assertRaisesRegex(asyncio.LimitOverrunError, 'is found') as cm: self.loop.run_until_complete(stream.readuntil(b'AAA')) self.assertEqual(b'some dataAAA', stream._buffer)
def parse_frame(reader: StreamReader, encoding: str, rabbitmq: bool) -> Frame: frame_type, channel_id, payload_size = unpack('!BHI', await reader.readexactly(7)) if frame_type == FRAME_TYPE_METHOD: frame = await parse_method_frame(reader, payload_size, channel_id, encoding, rabbitmq) elif frame_type == FRAME_TYPE_BODY: frame = BodyFrame(channel_id, payload_size, await reader.readexactly(payload_size)) elif frame_type == FRAME_TYPE_HEADER: frame = await parse_header_frame(reader, payload_size, channel_id, encoding, rabbitmq) elif frame_type == FRAME_TYPE_HEARTBEAT: frame = HEARTBEAT_FRAME else: raise InvalidFrameType(frame_type) frame_end = await reader.readexactly(1) if frame_end != FRAME_END: raise InvalidFrameEnd(frame_end) return frame
def test_020_get_events_reader_local(self): self.app.qubesd_connection_type = 'socket' loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) sock1, sock2 = socket.socketpair() with unittest.mock.patch('asyncio.open_unix_connection', lambda path: self.mock_open_unix_connection( qubesadmin.config.QUBESD_SOCKET, sock1, path)): task = asyncio.ensure_future(self.dispatcher._get_events_reader()) reader = asyncio.ensure_future(loop.run_in_executor(None, self.read_all, sock2)) loop.run_until_complete(asyncio.wait([task, reader])) self.assertEqual(reader.result(), b'dom0\0admin.Events\0dom0\0\0') self.assertIsInstance(task.result()[0], asyncio.StreamReader) cleanup_func = task.result()[1] cleanup_func() sock2.close() # run socket cleanup functions loop.stop() loop.run_forever() loop.close()
def test_021_get_events_reader_local_vm(self): self.app.qubesd_connection_type = 'socket' loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) sock1, sock2 = socket.socketpair() vm = unittest.mock.Mock() vm.name = 'test-vm' with unittest.mock.patch('asyncio.open_unix_connection', lambda path: self.mock_open_unix_connection( qubesadmin.config.QUBESD_SOCKET, sock1, path)): task = asyncio.ensure_future(self.dispatcher._get_events_reader(vm)) reader = asyncio.ensure_future(loop.run_in_executor(None, self.read_all, sock2)) loop.run_until_complete(asyncio.wait([task, reader])) self.assertEqual(reader.result(), b'dom0\0admin.Events\0test-vm\0\0') self.assertIsInstance(task.result()[0], asyncio.StreamReader) cleanup_func = task.result()[1] cleanup_func() sock2.close() # run socket cleanup functions loop.stop() loop.run_forever() loop.close()
def connection_made(self, transport): print("Got a client!") self.transport = transport # StreamReader is super convenient here; it has a regular method on our # end (feed_data), and a coroutine on the other end that will # faux-block until there's data to be read. We could also just call a # method directly on the screen, but this keeps the screen somewhat # separate from the protocol. self.reader = asyncio.StreamReader(loop=loop) screen = AsyncScreen(self.reader, transport) main_widget = build_widgets() self.urwid_loop = urwid.MainLoop( main_widget, event_loop=urwid.AsyncioEventLoop(loop=loop), screen=screen, unhandled_input=unhandled, ) self.urwid_loop.start()
def test_readline(self): # Read one line. 'readline' will need to wait for the data # to come from 'cb' stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(b'chunk1 ') read_task = asyncio.Task(stream.readline(), loop=self.loop) def cb(): stream.feed_data(b'chunk2 ') stream.feed_data(b'chunk3 ') stream.feed_data(b'\n chunk4') self.loop.call_soon(cb) line = self.loop.run_until_complete(read_task) self.assertEqual(b'chunk1 chunk2 chunk3 \n', line) self.assertEqual(b' chunk4', stream._buffer)
def __init__(self, config: JobConfig, retry_state: Optional[JobRetryState]) -> None: self.config = config self.proc = None # type: Optional[asyncio.subprocess.Process] self.retcode = None # type: Optional[int] self._stderr_reader = None # type: Optional[StreamReader] self._stdout_reader = None # type: Optional[StreamReader] self.stderr = None # type: Optional[str] self.stdout = None # type: Optional[str] self.stderr_discarded = 0 self.stdout_discarded = 0 self.execution_deadline = None # type: Optional[float] self.retry_state = retry_state self.env = None # type: Optional[Dict[str, str]] statsd_config = self.config.statsd if statsd_config is not None: self.statsd_writer = StatsdJobMetricWriter( host=statsd_config['host'], port=statsd_config['port'], prefix=statsd_config['prefix'], job=self, ) # type: Optional[StatsdJobMetricWriter] else: self.statsd_writer = None
def run(self): loop = asyncio.get_event_loop() age = None ppid = os.getpid() sockets = [] self.wsgi, app = make_stub_application(headers=self.response_headers, body=self.response_body) timeout = None cfg = Config() log = None sockname = ('127.0.0.1', '80') reader = asyncio.StreamReader() def feeder(): reader.feed_data(self.request) reader.feed_eof() worker = AsyncioWorker(age, ppid, sockets, app, timeout, cfg, log) loop.create_task(worker.connection_task(sockname, reader, self.writer)) loop.call_soon(feeder) run_worker(worker)
def get_reader(self): if self._reader is None: self._reader = asyncio.StreamReader() protocol = asyncio.StreamReaderProtocol(self._reader) loop = asyncio.get_event_loop() await loop.connect_read_pipe(lambda: protocol, sys.stdin) return self._reader
def stream(loop): return asyncio.StreamReader(loop=loop)
def connect_read_pipe(file): loop = asyncio.get_event_loop() stream_reader = asyncio.StreamReader(loop=loop) def factory(): return asyncio.StreamReaderProtocol(stream_reader) transport, _ = yield from loop.connect_read_pipe(factory, file) return stream_reader, transport # # Example #
def task(): rfd, wfd = os.pipe() args = [sys.executable, '-c', code, str(wfd)] pipe = open(rfd, 'rb', 0) reader = asyncio.StreamReader(loop=loop) protocol = asyncio.StreamReaderProtocol(reader, loop=loop) transport, _ = yield from loop.connect_read_pipe(lambda: protocol, pipe) proc = yield from asyncio.create_subprocess_exec(*args, pass_fds={wfd}) yield from proc.wait() os.close(wfd) data = yield from reader.read() print("read = %r" % data.decode())
def test_ctor_global_loop(self, m_events): stream = asyncio.StreamReader() self.assertIs(stream._loop, m_events.get_event_loop.return_value)
def test_feed_empty_data(self): stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(b'') self.assertEqual(b'', stream._buffer)
def test_feed_nonempty_data(self): stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(self.DATA) self.assertEqual(self.DATA, stream._buffer)
def test_read(self): # Read bytes. stream = asyncio.StreamReader(loop=self.loop) read_task = asyncio.Task(stream.read(30), loop=self.loop) def cb(): stream.feed_data(self.DATA) self.loop.call_soon(cb) data = self.loop.run_until_complete(read_task) self.assertEqual(self.DATA, data) self.assertEqual(b'', stream._buffer)
def test_read_line_breaks(self): # Read bytes without line breaks. stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(b'line1') stream.feed_data(b'line2') data = self.loop.run_until_complete(stream.read(5)) self.assertEqual(b'line1', data) self.assertEqual(b'line2', stream._buffer)
def test_read_eof(self): # Read bytes, stop at eof. stream = asyncio.StreamReader(loop=self.loop) read_task = asyncio.Task(stream.read(1024), loop=self.loop) def cb(): stream.feed_eof() self.loop.call_soon(cb) data = self.loop.run_until_complete(read_task) self.assertEqual(b'', data) self.assertEqual(b'', stream._buffer)
def test_read_until_eof(self): # Read all bytes until eof. stream = asyncio.StreamReader(loop=self.loop) read_task = asyncio.Task(stream.read(-1), loop=self.loop) def cb(): stream.feed_data(b'chunk1\n') stream.feed_data(b'chunk2') stream.feed_eof() self.loop.call_soon(cb) data = self.loop.run_until_complete(read_task) self.assertEqual(b'chunk1\nchunk2', data) self.assertEqual(b'', stream._buffer)
def test_read_exception(self): stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(b'line\n') data = self.loop.run_until_complete(stream.read(2)) self.assertEqual(b'li', data) stream.set_exception(ValueError()) self.assertRaises( ValueError, self.loop.run_until_complete, stream.read(2))
def test_readline_limit_with_existing_data(self): # Read one line. The data is in StreamReader's buffer # before the event loop is run. stream = asyncio.StreamReader(limit=3, loop=self.loop) stream.feed_data(b'li') stream.feed_data(b'ne1\nline2\n') self.assertRaises( ValueError, self.loop.run_until_complete, stream.readline()) # The buffer should contain the remaining data after exception self.assertEqual(b'line2\n', stream._buffer) stream = asyncio.StreamReader(limit=3, loop=self.loop) stream.feed_data(b'li') stream.feed_data(b'ne1') stream.feed_data(b'li') self.assertRaises( ValueError, self.loop.run_until_complete, stream.readline()) # No b'\n' at the end. The 'limit' is set to 3. So before # waiting for the new data in buffer, 'readline' will consume # the entire buffer, and since the length of the consumed data # is more than 3, it will raise a ValueError. The buffer is # expected to be empty now. self.assertEqual(b'', stream._buffer)
def test_at_eof(self): stream = asyncio.StreamReader(loop=self.loop) self.assertFalse(stream.at_eof()) stream.feed_data(b'some data\n') self.assertFalse(stream.at_eof()) self.loop.run_until_complete(stream.readline()) self.assertFalse(stream.at_eof()) stream.feed_data(b'some data\n') stream.feed_eof() self.loop.run_until_complete(stream.readline()) self.assertTrue(stream.at_eof())
def test_readline_limit(self): # Read one line. StreamReaders are fed with data after # their 'readline' methods are called. stream = asyncio.StreamReader(limit=7, loop=self.loop) def cb(): stream.feed_data(b'chunk1') stream.feed_data(b'chunk2') stream.feed_data(b'chunk3\n') stream.feed_eof() self.loop.call_soon(cb) self.assertRaises( ValueError, self.loop.run_until_complete, stream.readline()) # The buffer had just one line of data, and after raising # a ValueError it should be empty. self.assertEqual(b'', stream._buffer) stream = asyncio.StreamReader(limit=7, loop=self.loop) def cb(): stream.feed_data(b'chunk1') stream.feed_data(b'chunk2\n') stream.feed_data(b'chunk3\n') stream.feed_eof() self.loop.call_soon(cb) self.assertRaises( ValueError, self.loop.run_until_complete, stream.readline()) self.assertEqual(b'chunk3\n', stream._buffer)
def test_readline_nolimit_nowait(self): # All needed data for the first 'readline' call will be # in the buffer. stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(self.DATA[:6]) stream.feed_data(self.DATA[6:]) line = self.loop.run_until_complete(stream.readline()) self.assertEqual(b'line1\n', line) self.assertEqual(b'line2\nline3\n', stream._buffer)
def test_readline_eof(self): stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(b'some data') stream.feed_eof() line = self.loop.run_until_complete(stream.readline()) self.assertEqual(b'some data', line)
def test_readline_read_byte_count(self): stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(self.DATA) self.loop.run_until_complete(stream.readline()) data = self.loop.run_until_complete(stream.read(7)) self.assertEqual(b'line2\nl', data) self.assertEqual(b'ine3\n', stream._buffer)
def test_readline_exception(self): stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(b'line\n') data = self.loop.run_until_complete(stream.readline()) self.assertEqual(b'line\n', data) stream.set_exception(ValueError()) self.assertRaises( ValueError, self.loop.run_until_complete, stream.readline()) self.assertEqual(b'', stream._buffer)
def test_readexactly_zero_or_less(self): # Read exact number of bytes (zero or less). stream = asyncio.StreamReader(loop=self.loop) stream.feed_data(self.DATA) data = self.loop.run_until_complete(stream.readexactly(0)) self.assertEqual(b'', data) self.assertEqual(self.DATA, stream._buffer) data = self.loop.run_until_complete(stream.readexactly(-1)) self.assertEqual(b'', data) self.assertEqual(self.DATA, stream._buffer)
def test_exception(self): stream = asyncio.StreamReader(loop=self.loop) self.assertIsNone(stream.exception()) exc = ValueError() stream.set_exception(exc) self.assertIs(stream.exception(), exc)
def test_exception_waiter(self): stream = asyncio.StreamReader(loop=self.loop) @asyncio.coroutine def set_err(): stream.set_exception(ValueError()) t1 = asyncio.Task(stream.readline(), loop=self.loop) t2 = asyncio.Task(set_err(), loop=self.loop) self.loop.run_until_complete(asyncio.wait([t1, t2], loop=self.loop)) self.assertRaises(ValueError, t1.result)
def test_exception_cancel(self): stream = asyncio.StreamReader(loop=self.loop) t = asyncio.Task(stream.readline(), loop=self.loop) test_utils.run_briefly(self.loop) t.cancel() test_utils.run_briefly(self.loop) # The following line fails if set_exception() isn't careful. stream.set_exception(RuntimeError('message')) test_utils.run_briefly(self.loop) self.assertIs(stream._waiter, None)
def test_read_all_from_pipe_reader(self): # See Tulip issue 168. This test is derived from the example # subprocess_attach_read_pipe.py, but we configure the # StreamReader's limit so that twice it is less than the size # of the data writter. Also we must explicitly attach a child # watcher to the event loop. code = """\ import os, sys fd = int(sys.argv[1]) os.write(fd, b'data') os.close(fd) """ rfd, wfd = os.pipe() args = [sys.executable, '-c', code, str(wfd)] pipe = open(rfd, 'rb', 0) reader = asyncio.StreamReader(loop=self.loop, limit=1) protocol = asyncio.StreamReaderProtocol(reader, loop=self.loop) transport, _ = self.loop.run_until_complete( self.loop.connect_read_pipe(lambda: protocol, pipe)) watcher = asyncio.SafeChildWatcher() watcher.attach_loop(self.loop) try: asyncio.set_child_watcher(watcher) create = asyncio.create_subprocess_exec(*args, pass_fds={wfd}, loop=self.loop) proc = self.loop.run_until_complete(create) self.loop.run_until_complete(proc.wait()) finally: asyncio.set_child_watcher(None) os.close(wfd) data = self.loop.run_until_complete(reader.read(-1)) self.assertEqual(data, b'data')
def test_streamreader_constructor(self): self.addCleanup(asyncio.set_event_loop, None) asyncio.set_event_loop(self.loop) # Tulip issue #184: Ensure that StreamReaderProtocol constructor # retrieves the current loop if the loop parameter is not set reader = asyncio.StreamReader() self.assertIs(reader._loop, self.loop)
def stdio(loop=None): if loop is None: loop = asyncio.get_event_loop() reader = asyncio.StreamReader() reader_protocol = asyncio.StreamReaderProtocol(reader) writer_transport, writer_protocol = await loop.connect_write_pipe( FlowControlMixin, os.fdopen(0, 'wb')) writer = StreamWriter(writer_transport, writer_protocol, None, loop) await loop.connect_read_pipe(lambda: reader_protocol, sys.stdin) return reader, writer
def read_pipe(file, size): loop = get_event_loop() reader = StreamReader() protocol = StreamReaderProtocol(reader) transport, _ = await loop.connect_read_pipe( lambda: protocol, fdopen(os_open(file, O_RDONLY | O_NONBLOCK))) chunks = list() while size > 0: chunk = await reader.read(size) if not chunk: break chunks.append(chunk) size -= len(chunk) transport.close() return b''.join(chunks)
def stdio(loop=None): if loop is None: loop = asyncio.get_event_loop() reader = asyncio.StreamReader() reader_protocol = asyncio.StreamReaderProtocol(reader) writer_transport, writer_protocol = await loop.connect_write_pipe(FlowControlMixin, os.fdopen(1, 'wb')) writer = StreamWriter(writer_transport, writer_protocol, None, loop) await loop.connect_read_pipe(lambda: reader_protocol, sys.stdin) return reader, writer
