18.5.5. Streams (coroutine based API) — Python documentation
18.5.5. Streams (coroutine based API)
Source code: :source:`Lib/asyncio/streams.py`
18.5.5.1. Stream functions
Note
The top-level functions in this module are meant as convenience wrappers only; there’s really nothing special there, and if they don’t do exactly what you want, feel free to copy their code.
18.5.5.2. StreamReader
- class asyncio.StreamReader(limit=_DEFAULT_LIMIT, loop=None)
This class is not thread safe.
The limit argument’s default value is set to _DEFAULT_LIMIT which is 2**16 (64 KiB)
- exception()
Get the exception.
- feed_eof()
Acknowledge the EOF.
- feed_data(data)
Feed data bytes in the internal buffer. Any operations waiting for the data will be resumed.
- set_exception(exc)
Set the exception.
- set_transport(transport)
Set the transport.
- at_eof()
Return
True
if the buffer is empty and feed_eof() was called.
18.5.5.3. StreamWriter
- class asyncio.StreamWriter(transport, protocol, reader, loop)
Wraps a Transport.
This exposes write(), writelines(), can_write_eof(), 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 attribute which references theTransport
directly.This class is not thread safe.
- transport
Transport.
- can_write_eof()
Return True if the transport supports write_eof(), False if not. See WriteTransport.can_write_eof().
- close()
Close the transport: see BaseTransport.close().
- get_extra_info(name, default=None)
Return optional transport information: see BaseTransport.get_extra_info().
- write(data)
Write some data bytes to the transport: see WriteTransport.write().
- writelines(data)
Write a list (or any iterable) of data bytes to the transport: see WriteTransport.writelines().
- write_eof()
Close the write end of the transport after flushing buffered data: see WriteTransport.write_eof().
18.5.5.4. StreamReaderProtocol
- class asyncio.StreamReaderProtocol(stream_reader, client_connected_cb=None, loop=None)
Trivial helper class to adapt between Protocol and StreamReader. Subclass of Protocol.
stream_reader is a StreamReader instance, client_connected_cb is an optional function called with (stream_reader, stream_writer) when a connection is made, loop is the event loop instance to use.
(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 from accidentally calling inappropriate methods of the protocol.)
18.5.5.5. IncompleteReadError
- exception asyncio.IncompleteReadError
Incomplete read error, subclass of EOFError.
- expected
Total number of expected bytes (int).
- partial
Read bytes string before the end of stream was reached (bytes).
18.5.5.6. LimitOverrunError
- exception asyncio.LimitOverrunError
- Reached the buffer limit while looking for a separator.
- consumed
- Total number of to be consumed bytes.
18.5.5.7. Stream examples
18.5.5.7.1. TCP echo client using streams
TCP echo client using the asyncio.open_connection()
function:
import asyncio
@asyncio.coroutine
def tcp_echo_client(message, loop):
reader, writer = yield from asyncio.open_connection('127.0.0.1', 8888,
loop=loop)
print('Send: %r' % message)
writer.write(message.encode())
data = yield from reader.read(100)
print('Received: %r' % data.decode())
print('Close the socket')
writer.close()
message = 'Hello World!'
loop = asyncio.get_event_loop()
loop.run_until_complete(tcp_echo_client(message, loop))
loop.close()
See also
The TCP echo client protocol example uses the AbstractEventLoop.create_connection()
method.
18.5.5.7.2. TCP echo server using streams
TCP echo server using the asyncio.start_server()
function:
import asyncio
@asyncio.coroutine
def handle_echo(reader, writer):
data = yield from reader.read(100)
message = data.decode()
addr = writer.get_extra_info('peername')
print("Received %r from %r" % (message, addr))
print("Send: %r" % message)
writer.write(data)
yield from writer.drain()
print("Close the client socket")
writer.close()
loop = asyncio.get_event_loop()
coro = asyncio.start_server(handle_echo, '127.0.0.1', 8888, loop=loop)
server = loop.run_until_complete(coro)
# Serve requests until Ctrl+C is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
loop.run_forever()
except KeyboardInterrupt:
pass
# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()
18.5.5.7.3. Get HTTP headers
Simple example querying HTTP headers of the URL passed on the command line:
import asyncio
import urllib.parse
import sys
@asyncio.coroutine
def print_http_headers(url):
url = urllib.parse.urlsplit(url)
if url.scheme == 'https':
connect = asyncio.open_connection(url.hostname, 443, ssl=True)
else:
connect = asyncio.open_connection(url.hostname, 80)
reader, writer = yield from connect
query = ('HEAD {path} HTTP/1.0\r\n'
'Host: {hostname}\r\n'
'\r\n').format(path=url.path or '/', hostname=url.hostname)
writer.write(query.encode('latin-1'))
while True:
line = yield from reader.readline()
if not line:
break
line = line.decode('latin1').rstrip()
if line:
print('HTTP header> %s' % line)
# Ignore the body, close the socket
writer.close()
url = sys.argv[1]
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(print_http_headers(url))
loop.run_until_complete(task)
loop.close()
Usage:
python example.py http://example.com/path/page.html
or with HTTPS:
python example.py https://example.com/path/page.html
18.5.5.7.4. Register an open socket to wait for data using streams
Coroutine waiting until a socket receives data using the open_connection()
function:
import asyncio
try:
from socket import socketpair
except ImportError:
from asyncio.windows_utils import socketpair
@asyncio.coroutine
def wait_for_data(loop):
# Create a pair of connected sockets
rsock, wsock = socketpair()
# Register the open socket to wait for data
reader, writer = yield from asyncio.open_connection(sock=rsock, loop=loop)
# Simulate the reception of data from the network
loop.call_soon(wsock.send, 'abc'.encode())
# Wait for data
data = yield from reader.read(100)
# Got data, we are done: close the socket
print("Received:", data.decode())
writer.close()
# Close the second socket
wsock.close()
loop = asyncio.get_event_loop()
loop.run_until_complete(wait_for_data(loop))
loop.close()
See also
The register an open socket to wait for data using a protocol example uses a low-level protocol created by the AbstractEventLoop.create_connection()
method.
The watch a file descriptor for read events example uses the low-level AbstractEventLoop.add_reader() method to register the file descriptor of a socket.