transports.py
python
"""Абстрактный класс Transport."""
__all__ = (
'BaseTransport', 'ReadTransport', 'WriteTransport',
'Transport', 'DatagramTransport', 'SubprocessTransport',
)
class BaseTransport:
"""Базовый класс для транспортов."""
__slots__ = ('_extra',)
def __init__(self, extra=None):
if extra is None:
extra = {}
self._extra = extra
def get_extra_info(self, name, default=None):
"""Получить дополнительную информацию о транспорте."""
return self._extra.get(name, default)
def is_closing(self):
"""Возвращает True, если транспорт закрывается или закрыт."""
raise NotImplementedError
def close(self):
"""Закрывает транспорт.
Буферизованные данные будут асинхронно сброшены. Больше никаких данных
получено не будет. После того, как все буферизованные данные будут сброшены,
метод connection_lost() протокола будет (в конечном итоге) вызван
с None в качестве аргумента.
"""
raise NotImplementedError
def set_protocol(self, protocol):
"""Устанавливает новый протокол."""
raise NotImplementedError
def get_protocol(self):
"""Возвращает текущий протокол."""
raise NotImplementedError
class ReadTransport(BaseTransport):
"""Интерфейс для транспортов только для чтения."""
__slots__ = ()
def is_reading(self):
"""Возвращает True, если транспорт принимает данные."""
raise NotImplementedError
def pause_reading(self):
"""Приостановить приём.
Данные не будут передаваться методу data_received() протокола
до тех пор, пока не будет вызван resume_reading().
"""
raise NotImplementedError
def resume_reading(self):
"""Возобновить приём.
Полученные данные снова будут передаваться методу data_received()
протокола.
"""
raise NotImplementedError
class WriteTransport(BaseTransport):
"""Интерфейс для транспортов только для записи."""
__slots__ = ()
def set_write_buffer_limits(self, high=None, low=None):
"""Установить верхний и нижний пороги для управления потоком записи.
Эти два значения определяют, когда вызывать методы pause_writing()
и resume_writing() протокола. Если они заданы,
нижний порог должен быть меньше или равен
верхнему порогу. Ни одно из значений не может быть отрицательным.
Значения по умолчанию зависят от реализации. Если указан только
верхний порог, нижний порог по умолчанию устанавливается в
значение, зависящее от реализации, меньшее или равное
верхнему порогу. Установка верхнего порога в ноль также обнуляет
нижний порог и приводит к вызову pause_writing() каждый раз,
когда буфер становится непустым. Установка нижнего порога в ноль
приводит к вызову resume_writing() только после опустошения буфера.
Использование нуля для любого порога, как правило, неоптимально, так как
это уменьшает возможности для одновременного выполнения операций ввода-вывода
и вычислений.
"""
raise NotImplementedError
def get_write_buffer_size(self):
"""Вернуть текущий размер буфера записи."""
raise NotImplementedError
def get_write_buffer_limits(self):
"""Получает верхнюю и нижнюю границы для управления потоком записи.
Возвращает кортеж (low, high), где low и high — положительные числа в байтах."""
raise NotImplementedError
def write(self, data):
"""Записать некоторые байты данных в транспорт.
Этот вызов не блокируется; он буферизует данные и обеспечивает их
асинхронную отправку.
"""
raise NotImplementedError
def writelines(self, list_of_data):
"""Записывает список (или любую итерируемую последовательность) байтов данных в транспорт.
Реализация по умолчанию объединяет аргументы и
вызывает write() для результата.
"""
data = b''.join(list_of_data)
self.write(data)
def write_eof(self):
"""Закрывает конец записи после сброса буферизованных данных.
(Это похоже на ввод ^D в программу UNIX, читающую из stdin.)
Данные все еще могут приниматься.
"""
raise NotImplementedError
def can_write_eof(self):
"""Вернуть True, если этот транспорт поддерживает write_eof(), иначе False."""
raise NotImplementedError
def abort(self):
"""Закрыть транспорт немедленно.
Буферизованные данные будут потеряны. Новые данные приниматься не будут.
Метод connection_lost() протокола будет (в конечном итоге) вызван
с None в качестве аргумента.
"""
raise NotImplementedError
class Transport(ReadTransport, WriteTransport):
"""Интерфейс, представляющий двунаправленный транспорт.
Возможны несколько реализаций, но обычно пользователь не
реализует новые транспорты; вместо этого платформа предоставляет некоторые
полезные транспорты, которые реализованы с использованием лучших практик
платформы.
Пользователь никогда не создает экземпляр транспорта напрямую; он вызывает
служебную функцию, передавая ей фабрику протоколов и другую
информацию, необходимую для создания транспорта и протокола. (Например,
EventLoop.create_connection() или EventLoop.create_server().)
Служебная функция асинхронно создает транспорт и протокол
и соединяет их, вызывая метод connection_made() протокола,
передавая ему транспорт.
Реализация здесь вызывает NotImplemented для каждого метода,
кроме writelines(), который вызывает write() в цикле.
"""
__slots__ = ()
class DatagramTransport(BaseTransport):
"""Интерфейс для датаграммных (UDP) транспортов."""
__slots__ = ()
def sendto(self, data, addr=None):
"""Цикл событий Unix.
Добавляет поддержку обработки сигналов и сокетов домена UNIX в
SelectorEventLoop.
"""
raise NotImplementedError
def abort(self):
"""Закрыть транспорт немедленно.
Буферизованные данные будут потеряны. Новые данные приниматься не будут.
Метод connection_lost() протокола будет (в конечном итоге) вызван
с None в качестве аргумента.
"""
raise NotImplementedError
class SubprocessTransport(BaseTransport):
__slots__ = ()
def get_pid(self):
"""Получить идентификатор подпроцесса."""
raise NotImplementedError
def get_returncode(self):
"""Получить код возврата подпроцесса.
См. также
http://docs.python.org/3/library/subprocess#subprocess.Popen.returncode
"""
raise NotImplementedError
def get_pipe_transport(self, fd):
"""Получить транспорт для канала с номером fd."""
raise NotImplementedError
def send_signal(self, signal):
"""Отправить сигнал подпроцессу.
См. также:
docs.python.org/3/library/subprocess#subprocess.Popen.send_signal
"""
raise NotImplementedError
def terminate(self):
"""Остановить подпроцесс.
Псевдоним для метода close().
В ОС Posix метод отправляет SIGTERM подпроцессу.
В Windows вызывается функция Win32 API TerminateProcess()
для остановки подпроцесса.
См. также:
http://docs.python.org/3/library/subprocess#subprocess.Popen.terminate
"""
raise NotImplementedError
def kill(self):
"""Принудительно завершить подпроцесс.
В ОС Posix функция отправляет SIGKILL подпроцессу.
В Windows kill() является псевдонимом для terminate().
См. также:
http://docs.python.org/3/library/subprocess#subprocess.Popen.kill
"""
raise NotImplementedError
class _FlowControlMixin(Transport):
"""Вся логика управления потоком (записи) в базовом классе-примеси.
Подкласс должен реализовать get_write_buffer_size(). Он должен вызывать
_maybe_pause_protocol() при увеличении размера буфера записи,
и _maybe_resume_protocol() при его уменьшении. Он также может
переопределить set_write_buffer_limits() (например, чтобы указать другие
значения по умолчанию).
Конструктор подкласса должен вызывать super().__init__(extra). Это
вызовет set_write_buffer_limits().
Пользователь может вызывать set_write_buffer_limits() и
get_write_buffer_size(), а также могут быть вызваны pause_writing() и
resume_writing() их протокола.
"""
__slots__ = ('_loop', '_protocol_paused', '_high_water', '_low_water')
def __init__(self, extra=None, loop=None):
super().__init__(extra)
assert loop is not None
self._loop = loop
self._protocol_paused = False
self._set_write_buffer_limits()
def _maybe_pause_protocol(self):
size = self.get_write_buffer_size()
if size <= self._high_water:
return
if not self._protocol_paused:
self._protocol_paused = True
try:
self._protocol.pause_writing()
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._loop.call_exception_handler({
'message': 'protocol.pause_writing() failed',
'exception': exc,
'transport': self,
'protocol': self._protocol,
})
def _maybe_resume_protocol(self):
if (self._protocol_paused and
self.get_write_buffer_size() <= self._low_water):
self._protocol_paused = False
try:
self._protocol.resume_writing()
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._loop.call_exception_handler({
'message': 'protocol.resume_writing() failed',
'exception': exc,
'transport': self,
'protocol': self._protocol,
})
def get_write_buffer_limits(self):
return (self._low_water, self._high_water)
def _set_write_buffer_limits(self, high=None, low=None):
if high is None:
if low is None:
high = 64 * 1024
else:
high = 4 * low
if low is None:
low = high // 4
if not high >= low >= 0:
raise ValueError(
f'high ({high!r}) must be >= low ({low!r}) must be >= 0')
self._high_water = high
self._low_water = low
def set_write_buffer_limits(self, high=None, low=None):
self._set_write_buffer_limits(high=high, low=low)
self._maybe_pause_protocol()
def get_write_buffer_size(self):
raise NotImplementedError