protocols.py
python
"""Абстрактные базовые классы протокола."""
__all__ = (
'BaseProtocol', 'Protocol', 'DatagramProtocol',
'SubprocessProtocol', 'BufferedProtocol',
)
class BaseProtocol:
"""Общий базовый класс для интерфейсов протокола.
Обычно пользователь реализует протоколы, производные от BaseProtocol,
такие как Protocol или ProcessProtocol.
Единственный случай, когда BaseProtocol следует реализовывать напрямую — это
транспорт только для записи, например канал записи (write pipe)
"""
__slots__ = ()
def connection_made(self, transport):
"""Вызывается при установлении соединения.
Аргумент — это транспорт, представляющий соединение через канал.
Чтобы получать данные, ожидайте вызовов data_received().
Когда соединение закрыто, вызывается connection_lost().
"""
def connection_lost(self, exc):
"""Вызывается, когда соединение потеряно или закрыто.
Аргумент — объект исключения или None (последнее
означает, что получен обычный EOF или соединение было
прервано или закрыто).
"""
def pause_writing(self):
"""Вызывается, когда буфер транспорта превышает верхнюю отметку (high-water mark).
Вызовы pause и resume парные -- pause_writing() вызывается
один раз, когда буфер строго превышает верхнюю отметку
(даже если последующие записи ещё больше увеличивают размер буфера),
и в конечном итоге resume_writing() вызывается один раз, когда
размер буфера достигает нижней отметки (low-water mark).
Обратите внимание: если размер буфера равен верхней отметке,
pause_writing() не вызывается -- он должен быть строго больше.
И наоборот, resume_writing() вызывается, когда размер буфера
равен или меньше нижней отметки. Эти граничные условия
важны для обеспечения ожидаемого поведения, когда любая
отметка равна нулю.
ПРИМЕЧАНИЕ: Это единственный колбэк протокола, который не вызывается
через EventLoop.call_soon() -- если бы он вызывался, он бы не имел
эффекта, когда это наиболее необходимо (когда приложение продолжает
запись без уступки до тех пор, пока не будет вызван pause_writing()).
"""
def resume_writing(self):
"""Вызывается, когда буфер транспорта опускается ниже нижней отметки.
Подробнее см. в pause_writing().
"""
class Protocol(BaseProtocol):
"""Интерфейс для протокола потока данных.
Пользователь должен реализовать этот интерфейс. Он может наследовать от
этого класса, но это не обязательно. Реализации здесь ничего не делают
(они не вызывают исключений).
Когда пользователь хочет запросить транспорт, он передает фабрику
протоколов служебной функции (например, EventLoop.create_connection()).
Когда соединение успешно установлено, вызывается connection_made() с
подходящим объектом транспорта. Затем data_received() будет вызываться
0 или более раз с данными (bytes), полученными от транспорта;
наконец, connection_lost() будет вызван ровно один раз с объектом
исключения или None в качестве аргумента.
Конечный автомат вызовов:
start -> CM [-> DR*] [-> ER?] -> CL -> end
* CM: connection_made()
* DR: data_received()
* ER: eof_received()
* CL: connection_lost()
"""
__slots__ = ()
def data_received(self, data):
"""Вызывается при получении некоторых данных.
Аргумент — объект bytes.
"""
def eof_received(self):
"""Вызывается, когда удаленная сторона вызывает write_eof() или эквивалент.
Если этот метод возвращает ложное значение (включая None), транспорт
закроет себя сам. Если он возвращает истинное значение, закрытие
транспорта оставляется на усмотрение протокола.
"""
class BufferedProtocol(BaseProtocol):
"""Интерфейс для протокола потока данных с ручным управлением буфером.
Методы событий, такие как `create_server` и `create_connection`,
принимают фабрики, возвращающие протоколы, реализующие этот интерфейс.
Идея BufferedProtocol заключается в том, что он позволяет вручную выделять
и управлять приемным буфером. Циклы событий могут затем использовать буфер,
предоставленный протоколом, чтобы избежать ненужного копирования данных. Это
может привести к заметному улучшению производительности для протоколов,
которые получают большие объемы данных. Сложные протоколы могут выделять
буфер только один раз во время создания.
Конечный автомат вызовов:
start -> CM [-> GB [-> BU?]]* [-> ER?] -> CL -> end
* CM: connection_made()
* GB: get_buffer()
* BU: buffer_updated()
* ER: eof_received()
* CL: connection_lost()
"""
__slots__ = ()
def get_buffer(self, sizehint):
"""Вызывается для выделения нового приемного буфера.
*sizehint* — рекомендуемый минимальный размер возвращаемого
буфера. Если установлено -1, размер буфера может быть произвольным.
Должен возвращать объект, реализующий
:ref:`протокол буфера <bufferobjects>`.
Возврат буфера нулевого размера является ошибкой.
"""
def buffer_updated(self, nbytes):
"""Вызывается, когда буфер был обновлен полученными данными.
*nbytes* — общее количество байт, записанных в буфер.
"""
def eof_received(self):
"""Вызывается, когда удаленная сторона вызывает write_eof() или эквивалент.
Если этот метод возвращает ложное значение (включая None), транспорт
закроет себя сам. Если он возвращает истинное значение, закрытие
транспорта оставляется на усмотрение протокола.
"""
class DatagramProtocol(BaseProtocol):
"""Интерфейс протокола дейтаграмм."""
__slots__ = ()
def datagram_received(self, data, addr):
"""Вызывается при получении дейтаграммы."""
def error_received(self, exc):
"""Вызывается, когда операция отправки или получения вызывает OSError.
(За исключением BlockingIOError или InterruptedError.)
"""
class SubprocessProtocol(BaseProtocol):
"""Интерфейс протокола для вызовов подпроцессов."""
__slots__ = ()
def pipe_data_received(self, fd, data):
"""Вызывается, когда подпроцесс записывает данные в канал stdout/stderr.
fd — целочисленный файловый дескриптор.
data — объект bytes.
"""
def pipe_connection_lost(self, fd, exc):
"""Вызывается, когда файловый дескриптор, связанный с дочерним процессом,
закрывается.
fd — целочисленный файловый дескриптор, который был закрыт.
"""
def process_exited(self):
"""Вызывается, когда подпроцесс завершился."""
def _feed_data_to_buffered_proto(proto, data):
data_len = len(data)
while data_len:
buf = proto.get_buffer(data_len)
buf_len = len(buf)
if not buf_len:
raise RuntimeError('get_buffer() returned an empty buffer')
if buf_len >= data_len:
buf[:data_len] = data
proto.buffer_updated(data_len)
return
else:
buf[:buf_len] = data[:buf_len]
proto.buffer_updated(buf_len)
data = data[buf_len:]
data_len = len(data)