Транспорты и протоколы (API на основе обратных вызовов)

1. Транспорты

Транспорты — это классы, предоставляемые asyncio для абстрагирования различных типов каналов связи. Обычно вы не будете создавать транспорт самостоятельно; вместо этого вы вызовете метод AbstractEventLoop, который создаст транспорт и попытается инициировать базовый канал связи, вызвав обратный вызов при успехе.

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

asyncio в настоящее время реализует транспорты для TCP, UDP, SSL и труб подпроцессов. Доступные методы транспорта зависят от его типа.

Классы транспортов не являются потокобезопасными.

1.1. BaseTransport

class asyncio.BaseTransport

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

close()

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

is_closing()

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

Введено в версии 3.5.1.

get_extra_info(name, default=None)

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

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

• сокет:
  • '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

set_protocol(protocol)

Установить новый протокол. Переключение протоколов должно выполняться только в том случае, если оба протокола поддерживают такое переключение.

Введено в версии 3.5.3.

get_protocol()

Возвращает текущий протокол.

Введено в версии 3.5.3.

Изменено в версии 3.5.1: 'ssl_object' информация была добавлена для сокетов SSL.

1.2. ReadTransport

class asyncio.ReadTransport

Интерфейс для только-для-чтения транспортов.

pause_reading()

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

resume_reading()

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

1.3. WriteTransport

class asyncio.WriteTransport

Интерфейс для только-для-записи транспортов.

abort()

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

can_write_eof()

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

get_write_buffer_size()

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

get_write_buffer_limits()

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

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

Введено в версии 3.4.2.

set_write_buffer_limits(high=None, low=None)

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

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

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

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

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

write(data)

Записать данные data в транспорт.

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

writelines(list_of_data)

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

write_eof()

Закрыть конец записи транспорта после сброса данных из буфера. Данные могут по-прежнему приниматься.

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

1.4. DatagramTransport

DatagramTransport.sendto(data, addr=None)

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

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

DatagramTransport.abort()

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

1.5. BaseSubprocessTransport

class asyncio.BaseSubprocessTransport

get_pid()

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

get_pipe_transport(fd)

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

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

get_returncode()

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

kill()

Убить дочерний процесс, как в subprocess.Popen.kill().

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

send_signal(signal)

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

terminate()

Попросить дочерний процесс остановиться, как в subprocess.Popen.terminate(). Этот метод является псевдонимом для метода close().

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

close()

Попросить дочерний процесс остановиться, вызвав метод terminate(), если дочерний процесс еще не возвратил результат, и закрыть транспорты всех каналов (stdin, stdout и stderr).

2. Протоколы

asyncio предоставляет базовые классы, которые вы можете наследовать для реализации своих сетевых протоколов. Эти классы используются совместно с транспортами (см. ниже): протокол анализирует входящие данные и запрашивает запись исходящих данных, в то время как транспорт отвечает за фактические операции ввода-вывода и буферизацию.

При наследовании класса протокола рекомендуется переопределить определенные методы. Эти методы являются обработчиками: они будут вызываться транспортом при определенных событиях (например, при получении данных); не следует вызывать их самостоятельно, если вы не реализуете транспорт.

2.1. Классы протоколов

class asyncio.Protocol

Базовый класс для реализации протоколов потокового обмена (для использования с транспортом TCP и SSL).

class asyncio.DatagramProtocol

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

class asyncio.SubprocessProtocol

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

2.2. Обработчики соединений

Эти обработчики могут вызываться на экземплярах Protocol, DatagramProtocol и SubprocessProtocol:

BaseProtocol.connection_made(transport)

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

Аргумент transport — это транспорт, представляющий соединение. Вам необходимо сохранить его где-нибудь (например, в качестве атрибута), если вам это нужно.

BaseProtocol.connection_lost(exc)

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

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

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

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

SubprocessProtocol.pipe_data_received(fd, data)

Вызывается, когда дочерний процесс записывает данные в свой канал stdout или stderr. fd — целочисленный дескриптор файла канала. data — непустой объект bytes, содержащий данные.

SubprocessProtocol.pipe_connection_lost(fd, exc)

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

SubprocessProtocol.process_exited()

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

2.3. Протоколы потокового обмена

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

Protocol.data_received(data)

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

Примечание

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

Protocol.eof_received()

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

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

Примечание

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

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

Машина состояний:

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

Следующие обработчики вызываются на экземплярах DatagramProtocol instances.

DatagramProtocol.datagram_received(data, addr)

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

DatagramProtocol.error_received(exc)

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

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

2.5. Обработка потока

Эти обратные вызовы могут вызываться для экземпляров Protocol, DatagramProtocol и SubprocessProtocol:

BaseProtocol.pause_writing()

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

BaseProtocol.resume_writing()

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

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

2.6. Корутины и протоколы

Корутины могут быть запланированы в методе протокола с помощью ensure_future(), но порядок выполнения не гарантируется. Протоколы не осведомлены о корутинах, созданных в методах протокола, и поэтому не будут их ждать.

Для обеспечения надежного порядка выполнения используйте объекты потоков в корутине с yield from. Например, корутина StreamWriter.drain() может использоваться для ожидания, пока буфер записи не будет очищен.

3. Примеры протоколов

3.1. Протокол TCP-клиента эха

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

import asyncio

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

    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')
        print('Stop the event loop')
        self.loop.stop()

loop = asyncio.get_event_loop()
message = 'Hello World!'
coro = loop.create_connection(lambda: EchoClientProtocol(message, loop),
                              '127.0.0.1', 8888)
loop.run_until_complete(coro)
loop.run_forever()
loop.close()

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

3.2. Протокол TCP-сервера эха

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

import asyncio

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

loop = asyncio.get_event_loop()
# Each client connection will create a new protocol instance
coro = loop.create_server(EchoServerClientProtocol, '127.0.0.1', 8888)
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()

Transport.close() можно вызвать сразу после WriteTransport.write(), даже если данные еще не отправлены по сокету: оба метода асинхронны. yield from не требуется, так как эти методы транспорта не являются корутинами.

3.3. Протокол UDP-клиента эха

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

import asyncio

class EchoClientProtocol:
    def __init__(self, message, loop):
        self.message = message
        self.loop = loop
        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("Socket closed, stop the event loop")
        loop = asyncio.get_event_loop()
        loop.stop()

loop = asyncio.get_event_loop()
message = "Hello World!"
connect = loop.create_datagram_endpoint(
    lambda: EchoClientProtocol(message, loop),
    remote_addr=('127.0.0.1', 9999))
transport, protocol = loop.run_until_complete(connect)
loop.run_forever()
transport.close()
loop.close()

3.4. Протокол UDP-сервера эха

UDP-сервер эха, использующий метод AbstractEventLoop.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)

loop = asyncio.get_event_loop()
print("Starting UDP server")
# One protocol instance will be created to serve all client requests
listen = loop.create_datagram_endpoint(
    EchoServerProtocol, local_addr=('127.0.0.1', 9999))
transport, protocol = loop.run_until_complete(listen)

try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

transport.close()
loop.close()

3.5. Регистрация открытого сокета для ожидания данных с помощью протокола

Ожидание данных в сокете с использованием метода AbstractEventLoop.create_connection() с протоколом и последующее закрытие цикла событий

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

# Create a pair of connected sockets
rsock, wsock = socketpair()
loop = asyncio.get_event_loop()

class MyProtocol(asyncio.Protocol):
    transport = None

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

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

        # We are done: close the transport (it will call connection_lost())
        self.transport.close()

    def connection_lost(self, exc):
        # The socket has been closed, stop the event loop
        loop.stop()

# Register the socket to wait for data
connect_coro = loop.create_connection(MyProtocol, sock=rsock)
transport, protocol = loop.run_until_complete(connect_coro)

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

# Run the event loop
loop.run_forever()

# We are done, close sockets and the event loop
rsock.close()
wsock.close()
loop.close()