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.

Аргументы совпадают с обычными аргументами AbstractEventLoop.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 может быть как обычной функцией-колбэком, так и корутинной функцией; если это корутинная функция, она будет автоматически преобразована в Task.

Остальные аргументы – это все обычные аргументы 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=_DEFAULT_LIMIT, loop=None)¶

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

Значением по умолчанию аргумента limit является _DEFAULT_LIMIT, равный 2**16 (64 КиБ).

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 исключения содержит частично прочитанные байты.

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

coroutine readuntil(separator=b'\n')¶

Читает данные из потока, пока не будет найден separator.

В случае успеха данные и разделитель удаляются из внутреннего буфера (потребляются). Возвращаемые данные включают разделитель в конце.

Настроенный лимит потока используется для проверки результата. Лимит определяет максимальную длину возвращаемых данных, не включая разделитель.

Если возникает EOF, а полный разделитель всё ещё не найден, будет вызвано исключение IncompleteReadError, и внутренний буфер будет сброшен. Атрибут IncompleteReadError.partial может содержать часть разделителя.

Если данные не могут быть прочитаны из-за превышения лимита, будет вызвано исключение LimitOverrunError, а данные останутся во внутреннем буфере, чтобы их можно было прочитать снова.

Новое в версии 3.5.2.

at_eof()¶

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

18.5.5.3. StreamWriter¶

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

Оборачивает Transport.

Этот класс предоставляет write(), writelines(), can_write_eof(), write_eof(), get_extra_info() и close(). Он добавляет drain(), который возвращает необязательный Future, на котором можно ожидать управления потоком. Также добавляется атрибут транспорта, который ссылается непосредственно на Transport.

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

transport¶

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() даёт циклу событий возможность запланировать операцию записи и сбросить буфер. Это следует использовать особенно тогда, когда в транспорт записывается, возможно, большой объём данных, и корутина не передаёт управление между вызовами write().

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

get_extra_info(name, default=None)¶

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

write(data)¶

Записать некоторые 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.

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

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

18.5.5.5. IncompleteReadError¶

exception asyncio.IncompleteReadError¶

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

expected¶

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

partial¶

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

18.5.5.6. LimitOverrunError¶

exception asyncio.LimitOverrunError¶

Достигнут предел буфера во время поиска разделителя.

consumed¶

Общее количество байтов для обработки.

18.5.5.7. Примеры потоков¶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()