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

Исходный код: Lib/asyncio/transports.py

Исходный код: Lib/asyncio/protocols.py

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

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

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

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

Классы транспорта не потокобезопасны.

18.5.4.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.

18.5.4.1.2. ReadTransport

class asyncio.ReadTransport

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

pause_reading()

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

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

resume_reading()

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

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

18.5.4.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) не поддерживает полузакрытия.

18.5.4.1.4. DatagramTransport

DatagramTransport.sendto(data, addr=None)

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

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

DatagramTransport.abort()

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

18.5.4.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).

18.5.4.2. Протоколы

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

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

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

class asyncio.Protocol

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

class asyncio.DatagramProtocol

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

class asyncio.SubprocessProtocol

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

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

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

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

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

Protocol.data_received(data)

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

Примечание

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

Protocol.eof_received()

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

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

Примечание

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

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

Схема состояний:

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

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

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

DatagramProtocol.datagram_received(data, addr)

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

DatagramProtocol.error_received(exc)

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

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

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

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

BaseProtocol.pause_writing()

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

BaseProtocol.resume_writing()

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

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

18.5.4.2.6. Подпрограммы и протоколы

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

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

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

18.5.4.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(), цикл больше не работает, поэтому нет необходимости останавливать цикл в случае ошибки.

18.5.4.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 не требуется, так как эти методы транспорта не являются корутинами.

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

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

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