18.5.5. Потоки данных (API на основе корутин)¶Streams (coroutine based API)

18.5.5.1. Функции потоков данных¶Stream functions

coroutine asyncio.open_connection(host=None, port=None, *, loop=None, limit=None, **kwds)¶

Обёртка для create_connection(), возвращающая пару (reader, writer).

Возвращаемый reader – это экземпляр StreamReader, а writer – экземпляр StreamWriter.

Аргументы – это все обычные аргументы BaseEventLoop.create_connection(), кроме protocol_factory; наиболее распространены позиционные host и port, за которыми следуют различные необязательные ключевые аргументы.

Дополнительные необязательные ключевые аргументы: loop (для указания используемого экземпляра цикла событий) и limit (для задания лимита буфера, передаваемого в StreamReader).

Эта функция – корутина.

coroutine asyncio.start_server(client_connected_cb, host=None, port=None, *, loop=None, limit=None, **kwds)¶

Запускает сокетный сервер с колбэком для каждого подключившегося клиента. Возвращаемое значение такое же, как у create_server().

Параметр client_connected_cb вызывается с двумя параметрами: client_reader и client_writer. client_reader – это объект StreamReader, а client_writer – объект StreamWriter. Параметр client_connected_cb может быть либо обычной функцией колбэка, либо корутинной функцией; если это корутинная функция, она будет автоматически преобразована в задачу.

Остальные аргументы – это все обычные аргументы create_server(), кроме protocol_factory; наиболее распространены позиционные host и port, за которыми следуют различные необязательные ключевые аргументы.

Дополнительные необязательные ключевые аргументы: loop (для указания используемого экземпляра цикла событий) и limit (для задания лимита буфера, передаваемого в StreamReader).

Эта функция – корутина.

coroutine asyncio.open_unix_connection(path=None, *, loop=None, limit=None, **kwds)¶

Обёртка для create_unix_connection(), возвращающая пару (reader, writer).

См. open_connection() для получения информации о возвращаемом значении и других деталях.

Эта функция – корутина.

Доступность: UNIX.

coroutine asyncio.start_unix_server(client_connected_cb, path=None, *, loop=None, limit=None, **kwds)¶

Запускает сервер UNIX Domain Socket с колбэком для каждого подключившегося клиента.

См. start_server() для получения информации о возвращаемом значении и других деталях.

Эта функция – корутина.

Доступность: UNIX.

18.5.5.2. StreamReader¶

class asyncio.StreamReader(limit=None, loop=None)¶

This class is не является потокобезопасным.

exception()¶

Получить исключение.

feed_eof()¶

Подтверждает конец файла (EOF).

feed_data(data)¶

Подаёт байты data во внутренний буфер. Любые операции, ожидающие данные, будут возобновлены.

set_exception(exc)¶

Устанавливает исключение.

set_transport(transport)¶

Устанавливает транспорт.

coroutine read(n=-1)¶

Читает до n байт. Если n не указан или равен -1, читает до EOF и возвращает все прочитанные байты.

Если получен EOF и внутренний буфер пуст, возвращает пустой объект bytes.

Этот метод является корутиной.

coroutine readline()¶

Читает одну строку, где «строка» – это последовательность байтов, заканчивающаяся \n.

Если получен EOF, а \n не найден, метод вернёт частично прочитанные байты.

Если получен EOF и внутренний буфер пуст, возвращает пустой объект bytes.

Этот метод является корутиной.

coroutine readexactly(n)¶

Читает ровно n байт. Вызывает исключение IncompleteReadError, если конец потока достигнут до того, как прочитано n байт; атрибут IncompleteReadError.partial этого исключения содержит частично прочитанные байты.

Этот метод является корутиной.

at_eof()¶

Возвращает True, если буфер пуст и был вызван метод feed_eof().

18.5.5.3. StreamWriter¶

class asyncio.StreamWriter(transport, protocol, reader, loop)¶

Оборачивает 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 the Transport directly.

This class is не является потокобезопасным.

transport¶

Транспорт.

can_write_eof()¶

Возвращает True, если транспорт поддерживает write_eof(), и False в противном случае. См. WriteTransport.can_write_eof().

close()¶

Закрывает транспорт: см. BaseTransport.close().

coroutine drain()¶

Даёт возможность сбросить буфер записи нижележащего транспорта.

Предполагаемое использование – записать:

w.write(data)
yield from w.drain()

Когда размер буфера транспорта достигает верхнего порога (протокол приостанавливается), блокируется до тех пор, пока размер буфера не уменьшится до нижнего порога и протокол не будет возобновлён. Когда ждать нечего, yield-from продолжается немедленно.

Использование drain() с yield from даёт циклу событий возможность запланировать операцию записи и сбросить буфер. Особенно рекомендуется использовать этот метод, когда в транспорт записывается потенциально большой объём данных и корутина не использует yield from между вызовами write().

Этот метод является корутиной.

get_extra_info(name, default=None)¶

Возвращает дополнительную информацию о транспорте: см. BaseTransport.get_extra_info().

write(data)¶

Записывает данные байт в транспорт: см. WriteTransport.write().

writelines(data)¶

Записывает список (или любой итерируемый объект) байт данных в транспорт: см. WriteTransport.writelines().

write_eof()¶

Закрывает записывающий конец транспорта после сброса буферизованных данных: см. WriteTransport.write_eof().

18.5.5.4. StreamReaderProtocol¶

class asyncio.StreamReaderProtocol(stream_reader, client_connected_cb=None, loop=None)¶

Вспомогательный класс для адаптации между протоколом Protocol и StreamReader. Подкласс протокола Protocol.

потоковый_читатель – это экземпляр StreamReader, колбэк_подключения_клиента – необязательная функция, вызываемая с аргументами (stream_reader, stream_writer) при установке соединения, цикл_событий – экземпляр цикла событий, который следует использовать.

Это вспомогательный класс, а не превращение самого StreamReader в подкласс протокола Protocol, поскольку StreamReader имеет другие возможные применения, а также чтобы пользователь StreamReader случайно не вызывал неподходящие методы протокола.

18.5.5.5. IncompleteReadError¶

exception asyncio.IncompleteReadError¶

Ошибка неполного чтения, подкласс EOFError.

expected¶

Общее количество ожидаемых байт (int).

partial¶

Строка байтов прочитана до достижения конца потока (байты).

18.5.5.6. Примеры работы с потоками¶Stream examples

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()

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)

# Обслуживать запросы до нажатия Ctrl+C
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Закрыть сервер
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

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)

    # Игнорировать тело, закрыть сокет
    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()

python example.py http://example.com/path/page.html

python example.py https://example.com/path/page.html

import asyncio
try:
    from socket import socketpair
except ImportError:
    from asyncio.windows_utils import socketpair

@asyncio.coroutine
def wait_for_data(loop):
    # Создаём пару соединённых сокетов
    rsock, wsock = socketpair()

    # Зарегистрировать открытый сокет для ожидания данных
    reader, writer = yield from asyncio.open_connection(sock=rsock, loop=loop)

    # Симулировать приём данных из сети
    loop.call_soon(wsock.send, 'abc'.encode())

    # Ожидать данные
    data = yield from reader.read(100)

    # Данные получены, всё готово: закрыть сокет
    print("Received:", data.decode())
    writer.close()

    # Закрыть второй сокет
    wsock.close()

loop = asyncio.get_event_loop()
loop.run_until_complete(wait_for_data(loop))
loop.close()