18.5.4. Транспорты и протоколы (API на основе колбэков)¶Transports and protocols (callback based API)

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

resume_reading()¶

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

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

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

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

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

set_write_buffer_limits(high=None, low=None)¶

Устанавливает high- и low-пределы для управления потоком записи.

Эти два значения (измеряются в байтах) определяют, когда вызываются методы 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)¶

Записывает несколько байт данных в транспорт.

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

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

Отправляет подпроцессу номер 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.1. Классы протоколов¶Protocol classes

class asyncio.Protocol¶

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

class asyncio.DatagramProtocol¶

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

class asyncio.SubprocessProtocol¶

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

18.5.4.2.2. Колбэки соединения¶Connection callbacks

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

BaseProtocol.connection_made(transport)¶

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

Аргумент транспорт представляет соединение. Если он понадобится, его следует сохранить (например, как атрибут).

BaseProtocol.connection_lost(exc)¶

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

The argument is either an exception object or None. The latter means a regular EOF is received, or the connection was aborted or closed by this side of the connection.

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. Потоковые протоколы¶Streaming protocols

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

Protocol.data_received(data)¶

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

Примечание

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

Protocol.eof_received()¶

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

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

Примечание

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

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

Автомат состояний:

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

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

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

DatagramProtocol.datagram_received(data, addr)¶

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

DatagramProtocol.error_received(exc)¶

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

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

18.5.4.2.5. Колбэки управления потоком¶Flow control callbacks

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

BaseProtocol.pause_writing()¶

Вызывается, когда буфер транспорта превышает верхнюю границу (high-water mark).

BaseProtocol.resume_writing()¶

Вызывается, когда буфер транспорта опускается ниже нижней границы (low-water mark).

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

18.5.4.2.6. Корутины и протоколы¶Coroutines and protocols

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

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

18.5.4.3.1. Протокол TCP-клиента echo¶TCP echo client protocol

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-сервера echo¶TCP echo server protocol

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()
# Каждое клиентское соединение создаёт новый экземпляр протокола.
coro = loop.create_server(EchoServerClientProtocol, '127.0.0.1', 8888)
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()

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

18.5.4.3.3. Протокол UDP-эхо-клиента¶UDP echo client protocol

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 echo server protocol

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")
# Будет создан один экземпляр протокола для обслуживания всех клиентских запросов.
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. Регистрация открытого сокета для ожидания данных с помощью протокола¶Register an open socket to wait for data using a protocol

Ожидать, пока сокет получит данные, используя метод AbstractEventLoop.create_connection() с протоколом, а затем закрыть цикл событий

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

# Создаём пару соединённых сокетов
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())

        # Готово: закрыть транспорт (он вызовет connection_lost()).
        self.transport.close()

    def connection_lost(self, exc):
        # Сокет закрыт, остановить цикл событий.
        loop.stop()

# Зарегистрировать сокет для ожидания данных.
connect_coro = loop.create_connection(MyProtocol, sock=rsock)
transport, protocol = loop.run_until_complete(connect_coro)

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

# Запустить цикл событий
loop.run_forever()

# Готово, закрыть сокеты и цикл событий.
rsock.close()
wsock.close()
loop.close()