Транспорты и протоколы

Иерархия транспортов

class asyncio.BaseTransport

Базовый класс для всех транспортов. Содержит методы, общие для всех транспортов asyncio.

class asyncio.WriteTransport(BaseTransport)

Базовый транспорт для соединений только для записи.

Экземпляры класса WriteTransport возвращаются методом цикла событий loop.connect_write_pipe(), а также используются методами, связанными с подпроцессами, такими как loop.subprocess_exec().

class asyncio.ReadTransport(BaseTransport)

Базовый транспорт для соединений только для чтения.

Экземпляры класса ReadTransport возвращаются методом цикла событий loop.connect_read_pipe(), а также используются методами, связанными с подпроцессами, такими как loop.subprocess_exec().

class asyncio.Transport(WriteTransport, ReadTransport)

Интерфейс двунаправленного транспорта, например TCP-соединения.

Пользователь не создаёт транспорт напрямую; он вызывает вспомогательную функцию, передавая ей фабрику протоколов и другие сведения, необходимые для создания транспорта и протокола.

Экземпляры класса Transport возвращаются методами цикла событий или используются ими, например loop.create_connection(), loop.create_unix_connection(), loop.create_server(), loop.sendfile() и т. д.

class asyncio.DatagramTransport(BaseTransport)

Транспорт для дейтаграммных соединений (UDP).

Экземпляры класса DatagramTransport возвращаются методом цикла событий loop.create_datagram_endpoint().

class asyncio.SubprocessTransport(BaseTransport)

Абстракция, представляющая соединение между родительским процессом и его дочерним процессом ОС.

Экземпляры класса SubprocessTransport возвращаются методами цикла событий loop.subprocess_shell() и loop.subprocess_exec().

Базовый транспорт

BaseTransport.close()

Закрыть транспорт.

Если у транспорта есть буфер исходящих данных, данные из буфера будут асинхронно отправлены. Новые данные приниматься не будут. После отправки всех данных из буфера будет вызван метод протокола protocol.connection_lost(), которому в качестве аргумента будет передано None. После закрытия транспорт использовать не следует.

BaseTransport.is_closing()

Вернуть True, если транспорт закрывается или уже закрыт.

BaseTransport.get_extra_info(name, default=None)

Вернуть сведения о транспорте или используемых им базовых ресурсах.

name — строка, обозначающая тип запрашиваемых сведений о транспорте.

default — значение, возвращаемое, если сведения недоступны или если транспорт не поддерживает их запрос при использовании данной сторонней реализации цикла событий либо на текущей платформе.

Например, следующий код пытается получить объект базового сокета транспорта:

sock = transport.get_extra_info('socket')
if sock is not None:
    print(sock.getsockopt(...))

Категории сведений, которые можно запрашивать для некоторых транспортов:

• сокет:

  • 'peername': адрес удалённого узла, к которому подключён сокет; результат socket.socket.getpeername() (None в случае ошибки)
  • 'socket': экземпляр socket.socket
  • 'sockname': собственный адрес сокета; результат socket.socket.getsockname()

• SSL-сокет:

  • 'compression': используемый алгоритм сжатия в виде строки или None, если соединение не сжато; результат ssl.SSLSocket.compression()
  • 'cipher': кортеж из трёх значений, содержащий название используемого шифра, версию протокола SSL, определяющую его использование, и число используемых секретных битов; результат ssl.SSLSocket.cipher()
  • 'peercert': сертификат узла; результат ssl.SSLSocket.getpeercert()
  • 'sslcontext': экземпляр ssl.SSLContext
  • 'ssl_object': экземпляр ssl.SSLObject или ssl.SSLSocket

• канал:

  • 'pipe': объект канала

• подпроцесс:

  • 'subprocess': экземпляр subprocess.Popen

BaseTransport.set_protocol(protocol)

Установить новый протокол.

Переключать протокол следует только в том случае, если в документации обоих протоколов указана поддержка такого переключения.

BaseTransport.get_protocol()

Вернуть текущий протокол.

Транспорты только для чтения

ReadTransport.is_reading()

Вернуть True, если транспорт принимает новые данные.

Добавлено в версии 3.7.

ReadTransport.pause_reading()

Приостановить приём данных транспортом. Данные не будут передаваться методу протокола protocol.data_received() до вызова resume_reading().

Изменено в версии 3.7: Метод идемпотентен, то есть его можно вызывать, когда транспорт уже приостановлен или закрыт.

ReadTransport.resume_reading()

Возобновить приём данных. Метод протокола protocol.data_received() будет снова вызываться, если будут доступны данные для чтения.

Изменено в версии 3.7: Метод идемпотентен, то есть его можно вызывать, когда транспорт уже принимает данные.

Транспорты только для записи

WriteTransport.abort()

Немедленно закрыть транспорт, не дожидаясь завершения ожидающих операций. Данные из буфера будут потеряны. Новые данные приниматься не будут. В конечном итоге будет вызван метод протокола protocol.connection_lost(), которому в качестве аргумента будет передано None.

WriteTransport.can_write_eof()

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

WriteTransport.get_write_buffer_size()

Вернуть текущий размер выходного буфера, используемого транспортом.

WriteTransport.get_write_buffer_limits()

Получить верхний и нижний пороги управления потоком записи. Вернуть кортеж (low, high), где нижний и верхний пороги — положительные числа байтов.

Для установки порогов используйте set_write_buffer_limits().

Добавлено в версии 3.4.2.

WriteTransport.set_write_buffer_limits(high=None, low=None)

Установить верхний и нижний пороги управления потоком записи.

Эти два значения (измеряемые в байтах) определяют, когда вызываются методы протокола protocol.pause_writing() и protocol.resume_writing(). Если указан нижний порог, он должен быть меньше верхнего или равен ему. Ни верхний, ни нижний порог не могут быть отрицательными.

Метод pause_writing() вызывается, когда размер буфера становится равным верхнему порогу или превышает его. Если запись была приостановлена, метод resume_writing() вызывается, когда размер буфера становится равным нижнему порогу или оказывается меньше него.

Значения по умолчанию зависят от реализации. Если указан только верхний порог, нижний порог по умолчанию устанавливается в зависящее от реализации значение, меньшее или равное верхнему порогу. Если установить верхний порог в ноль, нижний порог также станет равен нулю, а метод pause_writing() будет вызываться при любом непустом буфере. Если установить нижний порог в ноль, метод resume_writing() будет вызываться только после опустошения буфера. Использование нуля для любого из порогов обычно не оптимально, поскольку оно сокращает возможности для одновременного выполнения операций ввода-вывода и вычислений.

Для получения порогов используйте get_write_buffer_limits().

WriteTransport.write(data)

Записать в транспорт байты data.

Этот метод не блокирует выполнение: он буферизует данные и организует их асинхронную отправку.

WriteTransport.writelines(list_of_data)

Записать в транспорт список (или любой итерируемый объект) байтов данных. По функциональности это эквивалентно вызову write() для каждого элемента, возвращаемого итератором, но может быть реализовано более эффективно.

WriteTransport.write_eof()

Закрыть конец транспорта для записи после отправки всех данных из буфера. Приём данных при этом может продолжаться.

Этот метод может вызвать исключение NotImplementedError, если транспорт (например, SSL) не поддерживает наполовину закрытые соединения.

Дейтаграммные транспорты

DatagramTransport.sendto(data, addr=None)

Отправить байты data удалённому узлу, указанному в addr (адрес назначения зависит от типа транспорта). Если addr равен None, данные отправляются по адресу назначения, заданному при создании транспорта.

Этот метод не блокирует выполнение: он буферизует данные и организует их асинхронную отправку.

DatagramTransport.abort()

Немедленно закрыть транспорт, не дожидаясь завершения ожидающих операций. Данные из буфера будут потеряны. Новые данные приниматься не будут. В конечном итоге будет вызван метод протокола protocol.connection_lost(), которому в качестве аргумента будет передано None.

Транспорты подпроцессов

SubprocessTransport.get_pid()

Вернуть идентификатор процесса подпроцесса в виде целого числа.

SubprocessTransport.get_pipe_transport(fd)

Вернуть транспорт канала связи, соответствующий целочисленному файловому дескриптору fd:

• 0: транспорт потокового чтения стандартного потока ввода (stdin) или None, если подпроцесс был создан без stdin=PIPE
• 1: транспорт потоковой записи стандартного потока вывода (stdout) или None, если подпроцесс был создан без stdout=PIPE
• 2: транспорт потоковой записи стандартного потока ошибок (stderr) или None, если подпроцесс был создан без stderr=PIPE
• другой fd: None

SubprocessTransport.get_returncode()

Вернуть код завершения подпроцесса в виде целого числа или None, если процесс ещё не завершился. Это аналогично атрибуту subprocess.Popen.returncode.

SubprocessTransport.kill()

Принудительно завершить подпроцесс.

В системах POSIX функция отправляет подпроцессу сигнал SIGKILL. В Windows этот метод является псевдонимом terminate().

См. также subprocess.Popen.kill().

SubprocessTransport.send_signal(signal)

Отправить подпроцессу сигнал с номером signal, как в subprocess.Popen.send_signal().

SubprocessTransport.terminate()

Остановить подпроцесс.

В системах POSIX этот метод отправляет подпроцессу сигнал SIGTERM. В Windows для остановки подпроцесса вызывается функция API Windows TerminateProcess().

См. также subprocess.Popen.terminate().

SubprocessTransport.close()

Завершить подпроцесс, вызвав метод kill().

Если подпроцесс ещё не завершился, закрыть транспорты каналов stdin, stdout и stderr.

Базовые протоколы

class asyncio.BaseProtocol

Базовый протокол с методами, общими для всех протоколов.

class asyncio.Protocol(BaseProtocol)

Базовый класс для реализации потоковых протоколов (TCP, сокеты Unix и т. д.).

class asyncio.BufferedProtocol(BaseProtocol)

Базовый класс для реализации потоковых протоколов с ручным управлением буфером приёма.

class asyncio.DatagramProtocol(BaseProtocol)

Базовый класс для реализации протоколов дейтаграмм (UDP).

class asyncio.SubprocessProtocol(BaseProtocol)

Базовый класс для реализации протоколов взаимодействия с дочерними процессами (однонаправленные каналы).

Базовый протокол

Все протоколы asyncio могут реализовывать обратные вызовы базового протокола.

Обратные вызовы соединения

Обратные вызовы соединения вызываются для всех протоколов ровно один раз при успешном установлении соединения. Все остальные обратные вызовы протокола могут вызываться только между этими двумя методами.

BaseProtocol.connection_made(transport)

Вызывается при установлении соединения.

Аргумент transport — это транспорт, представляющий соединение. Протокол отвечает за сохранение ссылки на свой транспорт.

BaseProtocol.connection_lost(exc)

Вызывается при потере или закрытии соединения.

Аргументом является объект исключения или None. Последнее означает, что получен обычный EOF либо соединение было прервано или закрыто этой стороной.

Обратные вызовы управления потоком

Транспорты могут вызывать обратные вызовы управления потоком, чтобы приостановить или возобновить запись, выполняемую протоколом.

Дополнительные сведения см. в документации метода set_write_buffer_limits().

BaseProtocol.pause_writing()

Вызывается, когда буфер транспорта превышает верхний порог.

BaseProtocol.resume_writing()

Вызывается, когда размер буфера транспорта становится меньше нижнего порога.

Если размер буфера равен верхнему порогу, pause_writing() не вызывается: размер буфера должен строго превысить этот порог.

И наоборот, resume_writing() вызывается, когда размер буфера становится равным нижнему порогу или меньше него. Эти граничные условия важны для того, чтобы обеспечить ожидаемое поведение, когда любой из порогов равен нулю.

Потоковые протоколы

Методы цикла событий, такие как loop.create_server(), loop.create_unix_server(), loop.create_connection(), loop.create_unix_connection(), loop.connect_accepted_socket(), loop.connect_read_pipe() и loop.connect_write_pipe(), принимают фабрики, возвращающие потоковые протоколы.

Protocol.data_received(data)

Вызывается при получении данных. data — непустой объект bytes, содержащий поступившие данные.

Будут ли данные буферизованы, разбиты на части или собраны заново, зависит от транспорта. В общем случае не следует полагаться на конкретную семантику; вместо этого следует сделать разбор данных универсальным и гибким. Однако данные всегда поступают в правильном порядке.

Метод может вызываться произвольное количество раз, пока соединение открыто.

Однако protocol.eof_received() вызывается не более одного раза. После вызова eof_received() метод data_received() больше не вызывается.

Protocol.eof_received()

Вызывается, когда другая сторона сообщает, что больше не будет отправлять данные (например, вызвав transport.write_eof(), если другая сторона тоже использует asyncio).

Этот метод может вернуть ложное значение (в том числе None); в этом случае транспорт закроется. Если же метод возвращает истинное значение, решение о закрытии транспорта зависит от используемого протокола. Поскольку реализация по умолчанию возвращает None, соединение закрывается неявно.

Некоторые транспорты, включая SSL, не поддерживают наполовину закрытые соединения; в этом случае возврат истинного значения из этого метода приведёт к закрытию соединения.

Конечный автомат:

start -> connection_made
    [-> data_received]*
    [-> eof_received]?
-> connection_lost -> end

Буферизованные потоковые протоколы

Буферизованные протоколы можно использовать с любым методом цикла событий, поддерживающим потоковые протоколы.

Реализации BufferedProtocol позволяют явно и вручную выделять буфер приёма и управлять им. Благодаря этому циклы событий могут использовать предоставленный протоколом буфер, избегая ненужного копирования данных. Это может заметно повысить производительность протоколов, принимающих большие объёмы данных. Сложные реализации протоколов могут значительно сократить количество выделений буфера.

Следующие обратные вызовы вызываются для экземпляров BufferedProtocol:

BufferedProtocol.get_buffer(sizehint)

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

sizehint — рекомендуемый минимальный размер возвращаемого буфера. Допускается возвращать буфер меньшего или большего размера, чем предлагает значение sizehint. Если значение равно -1, размер буфера может быть произвольным. Возврат буфера нулевого размера считается ошибкой.

Метод get_buffer() должен возвращать объект, реализующий буферный протокол.

BufferedProtocol.buffer_updated(nbytes)

Вызывается после обновления буфера полученными данными.

nbytes — общее количество байтов, записанных в буфер.

BufferedProtocol.eof_received()

См. документацию метода protocol.eof_received().

get_buffer() может вызываться произвольное количество раз в течение соединения. Однако protocol.eof_received() вызывается не более одного раза; если он был вызван, то get_buffer() и buffer_updated() больше не вызываются.

Конечный автомат:

start -> connection_made
    [-> get_buffer
        [-> buffer_updated]?
    ]*
    [-> eof_received]?
-> connection_lost -> end

Протоколы дейтаграмм

Экземпляры протоколов дейтаграмм следует создавать с помощью фабрик протоколов, передаваемых методу loop.create_datagram_endpoint().

DatagramProtocol.datagram_received(data, addr)

Вызывается при получении дейтаграммы. data — объект bytes, содержащий поступившие данные. addr — адрес узла, отправившего данные; точный формат зависит от транспорта.

DatagramProtocol.error_received(exc)

Вызывается, когда предыдущая операция отправки или получения вызывает исключение OSError. exc — экземпляр OSError.

Этот метод вызывается в редких случаях, когда транспорт (например, UDP) обнаруживает, что дейтаграмму не удалось доставить получателю. Однако во многих случаях недоставленные дейтаграммы отбрасываются без уведомления.

Протоколы подпроцессов

Экземпляры протоколов подпроцессов следует создавать с помощью фабрик протоколов, передаваемых методам loop.subprocess_exec() и loop.subprocess_shell().

SubprocessProtocol.pipe_data_received(fd, data)

Вызывается, когда дочерний процесс записывает данные в канал stdout или stderr.

fd — целочисленный файловый дескриптор канала.

data — непустой объект bytes, содержащий полученные данные.

SubprocessProtocol.pipe_connection_lost(fd, exc)

Вызывается при закрытии одного из каналов связи с дочерним процессом.

fd — целочисленный файловый дескриптор закрытого канала.

SubprocessProtocol.process_exited()

Вызывается после завершения дочернего процесса.

Может быть вызван до методов pipe_data_received() и pipe_connection_lost().

TCP-сервер эхо

Создайте TCP-сервер эхо с помощью метода loop.create_server(), отправляйте полученные данные обратно и закрывайте соединение:

import asyncio


class EchoServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        message = data.decode()
        print('Data received: {!r}'.format(message))

        print('Send: {!r}'.format(message))
        self.transport.write(data)

        print('Close the client socket')
        self.transport.close()


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    server = await loop.create_server(
        lambda: EchoServerProtocol(),
        '127.0.0.1', 8888)

    async with server:
        await server.serve_forever()


asyncio.run(main())

TCP-клиент эхо

TCP-клиент эхо с помощью метода loop.create_connection() отправляет данные и ждёт закрытия соединения:

import asyncio


class EchoClientProtocol(asyncio.Protocol):
    def __init__(self, message, on_con_lost):
        self.message = message
        self.on_con_lost = on_con_lost

    def connection_made(self, transport):
        transport.write(self.message.encode())
        print('Data sent: {!r}'.format(self.message))

    def data_received(self, data):
        print('Data received: {!r}'.format(data.decode()))

    def connection_lost(self, exc):
        print('The server closed the connection')
        self.on_con_lost.set_result(True)


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    on_con_lost = loop.create_future()
    message = 'Hello World!'

    transport, protocol = await loop.create_connection(
        lambda: EchoClientProtocol(message, on_con_lost),
        '127.0.0.1', 8888)

    # Wait until the protocol signals that the connection
    # is lost and close the transport.
    try:
        await on_con_lost
    finally:
        transport.close()


asyncio.run(main())

UDP-сервер эхо

UDP-сервер эхо с помощью метода loop.create_datagram_endpoint() отправляет полученные данные обратно:

import asyncio


class EchoServerProtocol:
    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        message = data.decode()
        print('Received %r from %s' % (message, addr))
        print('Send %r to %s' % (message, addr))
        self.transport.sendto(data, addr)


async def main():
    print("Starting UDP server")

    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    # One protocol instance will be created to serve all
    # client requests.
    transport, protocol = await loop.create_datagram_endpoint(
        lambda: EchoServerProtocol(),
        local_addr=('127.0.0.1', 9999))

    try:
        await asyncio.sleep(3600)  # Serve for 1 hour.
    finally:
        transport.close()


asyncio.run(main())

UDP-клиент эхо

UDP-клиент эхо с помощью метода loop.create_datagram_endpoint() отправляет данные и закрывает транспорт после получения ответа:

import asyncio


class EchoClientProtocol:
    def __init__(self, message, on_con_lost):
        self.message = message
        self.on_con_lost = on_con_lost
        self.transport = None

    def connection_made(self, transport):
        self.transport = transport
        print('Send:', self.message)
        self.transport.sendto(self.message.encode())

    def datagram_received(self, data, addr):
        print("Received:", data.decode())

        print("Close the socket")
        self.transport.close()

    def error_received(self, exc):
        print('Error received:', exc)

    def connection_lost(self, exc):
        print("Connection closed")
        self.on_con_lost.set_result(True)


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    on_con_lost = loop.create_future()
    message = "Hello World!"

    transport, protocol = await loop.create_datagram_endpoint(
        lambda: EchoClientProtocol(message, on_con_lost),
        remote_addr=('127.0.0.1', 9999))

    try:
        await on_con_lost
    finally:
        transport.close()


asyncio.run(main())

Подключение существующих сокетов

Дождитесь получения данных сокетом, используя метод loop.create_connection() с протоколом:

import asyncio
import socket


class MyProtocol(asyncio.Protocol):

    def __init__(self, on_con_lost):
        self.transport = None
        self.on_con_lost = on_con_lost

    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        print("Received:", data.decode())

        # We are done: close the transport;
        # connection_lost() will be called automatically.
        self.transport.close()

    def connection_lost(self, exc):
        # The socket has been closed
        self.on_con_lost.set_result(True)


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()
    on_con_lost = loop.create_future()

    # Create a pair of connected sockets
    rsock, wsock = socket.socketpair()

    # Register the socket to wait for data.
    transport, protocol = await loop.create_connection(
        lambda: MyProtocol(on_con_lost), sock=rsock)

    # Simulate the reception of data from the network.
    loop.call_soon(wsock.send, 'abc'.encode())

    try:
        await protocol.on_con_lost
    finally:
        transport.close()
        wsock.close()

asyncio.run(main())

loop.subprocess_exec() и SubprocessProtocol

Пример протокола подпроцесса, используемого для получения вывода подпроцесса и ожидания его завершения.

Подпроцесс создаётся методом loop.subprocess_exec():

import asyncio
import sys

class DateProtocol(asyncio.SubprocessProtocol):
    def __init__(self, exit_future):
        self.exit_future = exit_future
        self.output = bytearray()
        self.pipe_closed = False
        self.exited = False

    def pipe_connection_lost(self, fd, exc):
        self.pipe_closed = True
        self.check_for_exit()

    def pipe_data_received(self, fd, data):
        self.output.extend(data)

    def process_exited(self):
        self.exited = True
        # process_exited() method can be called before
        # pipe_connection_lost() method: wait until both methods are
        # called.
        self.check_for_exit()

    def check_for_exit(self):
        if self.pipe_closed and self.exited:
            self.exit_future.set_result(True)

async def get_date():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    code = 'import datetime; print(datetime.datetime.now())'
    exit_future = asyncio.Future(loop=loop)

    # Create the subprocess controlled by DateProtocol;
    # redirect the standard output into a pipe.
    transport, protocol = await loop.subprocess_exec(
        lambda: DateProtocol(exit_future),
        sys.executable, '-c', code,
        stdin=None, stderr=None)

    # Wait for the subprocess exit using the process_exited()
    # method of the protocol.
    await exit_future

    # Close the stdout pipe.
    transport.close()

    # Read the output which was collected by the
    # pipe_data_received() method of the protocol.
    data = bytes(protocol.output)
    return data.decode('ascii').rstrip()

date = asyncio.run(get_date())
print(f"Current date: {date}")

См. также тот же пример, написанный с использованием высокоуровневых API.