Пул соединений и трансляция

Соединения (WebSocket-сервера / SSE / WebSocket-клиентские исходящие) в ErisPulse являются первоклассными ресурсами на уровне фреймворка: при установке соединения автоматически регистрируются в единой таблице, поддерживают трансляцию, подписку на бизнес-группы и передачу/повторное использование между модулями, при выгрузке модуля автоматически закрываются и освобождаются фреймворком — бизнес-логика больше не должна создавать _ws_clients коллекции, и не будет утечек из-за забытого удаления.

Примеры в документации используют рекомендуемый синтаксис корневого импорта (рекомендуется с версии 2.10+): from ErisPulse import connections, router, client, WebSocketConnection

Обзор

from ErisPulse import connections, router, WebSocketConnection

# Сервер: соединение автоматически регистрируется, handler добавляет в бизнес-группу
@router.ws("MyModule", "/ws")
async def handle(ws: WebSocketConnection):
    ws.join_group("room:1")            # Название группы полностью определяется бизнес-логикой
    async for msg in ws.iter_text():
        ...                             # При разрыве соединения фреймворк автоматически отменяет регистрацию и очищает группу

# В любом месте: трансляция в группу (возвращает подробности успеха/неудачи)
result = await connections.broadcast({"type": "notify"}, group="room:1")
if not result.ok:
    ...                                 # result.failed: {id соединения: исключение}

Три вида соединений объединяются в единую таблицу регистрации:

Для отключения автоматической регистрации: передайте track=False в маршрут или при исходящем соединении, старый код будет работать без изменений.

Просмотр пула соединений

from ErisPulse import connections

# Просмотр пула соединений модуля ("Сколько активных соединений сейчас у модуля?")
conns = connections.list(namespace="Dashboard")
for conn in conns:
    print(conn.id, conn.kind, conn.groups, conn.meta)

# Фильтрация по другим параметрам (все условия AND)
connections.list(owner="MyAdapter", kind="client")
connections.list(group="tenant:acme")

# Общая статистика (подсчёт по всем измерениям)
stats = connections.stats()
# {'total': 5, 'by_kind': {...}, 'by_namespace': {...}, 'by_owner': {...}, 'groups': {...}}

# Подсчёт аудита принадлежности (модуль ownership также доступен)
from ErisPulse import ownership
ownership.counts("Dashboard")   # {'connections': 3, ...}

У каждого соединения можно получить идентификационную информацию: conn.id (вроде Dashboard:1a2b3c4d), conn.namespace, conn.owner, conn.kind, conn.groups, conn.meta (свободный словарь бизнес-логики).

Трансляция и подписка

Группы (модель подписки)

Группы — это плоские строки, их именование полностью определяется бизнес-логикой (комнаты / арендаторы / темы...), вложенные уровни выражаются соглашениями в именах (например tenant:acme/room:1), фреймворк не ограничивает формат.

Операции возможны в двух направлениях:

# Направление 1: соединение активно присоединяется/отключается (внутри handler)
@router.ws("Chat", "/ws")
async def handle(ws: WebSocketConnection):
    ws.join_group("room:1")        # Эквивалентно connections.assign(ws.id, "room:1")
    ws.leave_group("room:1")

# Направление 2: управление пассивно распределяет (внутри маршрута / других модулях)
connections.assign(conn_id, "tenant:acme")
connections.dismiss(conn_id, "tenant:acme")   # Без указания группы = выход из всех групп

При разрыве соединения автоматически выходят из всех групп, бизнес-логика не должна очищать.

Трансляция

result = await connections.broadcast(
    data,                       # Для WS используется send_json; для SSE — send (автоматическая сериализация JSON)
    namespace="Dashboard",      # Фильтр по пространству имён
    group="room:1",             # Фильтр по группе
    kind="sse",                 # Фильтр по типу соединения
    owner="MyAdapter",          # Фильтр по владельцу
    ids={conn_id, ...},         # Явное указание кандидатов (фильтры всё ещё действуют)
    exclude={conn_id},          # Исключение некоторых соединений
    timeout=10.0,               # Таймаут отправки на одно соединение (не общий)
    concurrency=64,             # Максимальное количество одновременных отправок
    raise_on_error=False,       # При True выбрасывается первое исключение при неудаче
)
result.total      # Общее количество целевых соединений (исключая предварительно)
result.sent       # Список успешно отправленных соединений
result.failed     # {id соединения: исключение} (таймаут/разрыв/ошибка отправки)
result.excluded   # Исключённые соединения
result.ok         # Успешна ли отправка всем

Фильтры между собой работают по правилу AND; неудача одного соединения не влияет на другие; соединения из ids, которых уже не существует, учитываются в failed (чтобы бизнес-логика знала полную картину).

События жизненного цикла

При регистрации/удалении соединения/изменении групп генерируются стандартные события жизненного цикла, которые могут подписываться Dashboard и другие, для отображения в реальном времени:

from ErisPulse import lifecycle

@lifecycle.on("connection.registered")
async def on_registered(data): ...
@lifecycle.on("connection.unregistered")
async def on_unregistered(data): ...
@lifecycle.on("connection.group.joined")
async def on_joined(data): ...
@lifecycle.on("connection.group.left")
async def on_left(data): ...

Передача / повторное использование между модулями

Соединение является зарегистрированным ресурсом фреймворка, любой модуль может получить его по ID и повторно использовать — не нужно создавать новое соединение, и не нужно знать, кто его создал:

from ErisPulse import connections

conn = connections.get("Dashboard:1a2b3c4d")   # Если не существует/разорвано выбрасывается ConnectionNotFoundError
await conn.send_json({"ping": 1})              # Любой модуль может отправить
conn.join_group("someone-elses-room")          # Также может участвовать в группах

Принадлежность и права закрытия:

• Владелец модуль, создавший соединение, имеет исключительное право закрытия. При вызове close() из не владельца выбрасывается ConnectionPermissionError (импорт from ErisPulse import ConnectionPermissionError);
• Операции отправки и присоединения доступны для всех модулей, не требуют авторизации;
• При выгрузке модуля / остановке адаптера фреймворк закрывает и отменяет регистрацию всех его соединений (серверных и исходящих), ссылки на уже полученные соединения при отправке получат результат неудачи, подписка на событие connection.unregistered позволит отслеживать;
• Внутренний путь освобождения фреймворка использует close(force=True) для обхода проверки.

Объяснение: проверка прав закрытия зависит от контекста владельца во время выполнения (например owner_scope / событийная диспетчеризация / spawn_background и т.д. автоматически передают). Вызовы без контекста принадлежности (старый код без контекста) разрешаются, чтобы не нарушить работу; новые вызовы в контексте владельца будут защищены.

Исходящие соединения (клиентская сторона)

from ErisPulse import client

ws = await client.ws_connect("wss://example.com/ws")
ws.id          # Уже зарегистрирован, доступен для поиска/трансляции через connections
ws.owner       # Текущий модуль (при недоступности контекста владельца рекомендуется явно указать)

# При недоступности контекста владельца (например, в обратном вызове из потока инструментов) явно указать, чтобы при выгрузке могло автоматически освободиться:
ws = await client.ws_connect("wss://example.com/ws", owner="MyAdapter")

При активном разрыве удалённой стороны, локальном close() или выгрузке модуля автоматически отменяется регистрация; с момента выгрузки старые способы создания и управления соединениями без регистрации и очистки устарели.

Передача между потоками

Для доступа к соединению из подпотока (трансляция / отправка / закрытие) используйте стандартные средства фреймворка для возврата в главный цикл, не используйте вручную run_coroutine_threadsafe:

from ErisPulse import run_main_loop, spawn_later, spawn_thread

# Блокирующий вызов из подпотока:
run_main_loop(connections.broadcast({"tick": 1}, group="room:1"), timeout=5)

# Фоновый поток + задержанная задача (принадлежность владельца, автоматически отменяется при выгрузке):
def push_tick():
    return connections.broadcast({"tick": 1}, group="room:1")
spawn_later(30, push_tick, owner="MyModule")

Семантика выгрузки и очистки

Цепочка возврата принадлежности при выгрузке модуля / остановке адаптера последовательно: отмена фоновых задач и таймеров принадлежащих владельцу → закрытие и отмена регистрации принадлежащих соединений → вызов хука on_cleanup → отмена регистрации маршрутов/обработчиков и других ресурсов. Подробнее см. Система принадлежности (owner).

Справочник API