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

18.5.4.1.1. BaseTransport¶

class asyncio.BaseTransport¶

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

close(self)¶

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

is_closing(self)¶

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

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

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': объект канала (pipe)
• подпроцесс:
  • 'подпроцесс': экземпляр subprocess.Popen

Изменено в версии 3.4.4: '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 не могут быть отрицательными.

Значения по умолчанию зависят от реализации. Если задан только верхний порог, нижний порог по умолчанию принимает значение, зависящее от реализации, которое меньше или равно верхнему порогу. Установка 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(self)¶

Завершает подпроцесс, как в 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)¶

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

Аргументом является либо объект исключения, либо 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. Потоковые протоколы¶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() вызывается один раз, когда буфер строго превышает верхнюю границу (high-water mark) (даже если последующие записи ещё больше увеличат размер буфера), и в итоге resume_writing() вызывается один раз, когда размер буфера достигает нижней границы (low-water mark).

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

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

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

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

TCP-эхо-клиент с использованием метода BaseEventLoop.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-эхо-сервер с использованием метода BaseEventLoop.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-эхо-клиент с использованием метода BaseEventLoop.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-эхо-сервер с использованием метода BaseEventLoop.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

Ожидание получения данных сокетом с помощью метода BaseEventLoop.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()