Tunnel — API

Клиент localzet\Tunnel\Client

Метод	Назначение
connect(?string $socketName = null): void	Создать одно статическое соединение, если его ещё нет. Передайте адрес при первом вызове.
on(string $event, callable $callback): void	Установить callback события и подписаться.
`subscribe(array	string $events): void`
`unsubscribe(array	string $events): void`
`publish(array	string $events, mixed $data, bool $isLoop = false): void`
`watch(array	string $channels, callable $callback, bool $autoReserve = true): void`
`unwatch(array	string $channels): void`
`enqueue(array	string $channels, mixed $data): void`
reserve(): void	Объявить готовность получить очередное задание по наблюдаемым очередям.
clearTimer(): void	Удалить внутренние ping/reconnect timers. Это не полный disconnect и не удаление подписок.

subscribe, unsubscribe, watch, unwatch и reserve используют send(), которое требует среды Server. publish/enqueue идут через отдельный sendAnyway, но standalone-ветка подключения содержит описанную в руководстве ошибку работы с resource.

Публичные callback-поля: $onMessage($channel, $data), $onConnect(), $onClose(). $pingInterval по умолчанию 25 секунд. Ping отправляет пустое сообщение, а не проверяет подтверждение сохранения данных сервером.

Событийный lifecycle

onConnect() повторяет event-subscriptions и очищает внутренние timers, в том числе ping-timer. Наличие публичного pingInterval не обещает, что heartbeat будет продолжаться после этой очистки. onClose() ставит reconnect через повторяющийся Timer с интервалом 1 секунда.

При приёме event callback получает только $data; общий onMessage получает $channel и $data. Queue callback получает $data. Событие и очередь с одинаковым именем имеют разные maps callbacks.

Envelope

Команды сериализуются как JSON либо PHP serialize в зависимости от флага соединения:

{"type":"publish","channels":["order.created"],"data":{"id":42}}

Тип	Дополнительные поля
subscribe, unsubscribe, watch, unwatch	channels: массив строк.
publish, publishLoop, enqueue	channels и data.
reserve	В клиентской команде channels не задаётся; сервер использует список watchs соединения.

Ответ событийной рассылки: {"type":"event","channel":"order.created","data":...}. Ответ очереди использует type = queue. На сервере обработчик до switch обращается к полю channels без проверки наличия; для reserve возможна PHP-диагностика отсутствующего ключа, хотя ветка использует watchs.

Формат payload и framing транспорта — разные слои. При выборе Frame TCP-данные должны иметь заголовок протокола Server; передача одного JSON без framing не является тем же socket protocol.

Сервер localzet\Tunnel\Server

Конструктор принимает ?string $socketName = null и bool $debug = false. При заданном адресе регистрирует handler через localzet_start() с count=1 и reloadable=false.

onServerStart() сохраняет ссылку на Server и создаёт карту каналов. onMessage() обрабатывает команды. onClose() удаляет event-subscriptions и watchers закрытого соединения, а пустые очереди без watchers удаляет из карты.

Внутренняя очередь

Queue использует SplQueue и две карты — watcher и consumer. addWatch/removeWatch меняют наблюдение, addConsumer добавляет готовность и вызывает dispatch, enqueue добавляет данные и тоже вызывает dispatch.

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