Tunnel

Tunnel передаёт сообщения между PHP-процессами через отдельный сервер. Он содержит два механизма: pub/sub для событий и очередь в памяти для выдачи заданий свободным потребителям.

Сервер не сохраняет сообщения на диск и не предоставляет аутентификацию клиентов. Работа с каналом не равнозначна защищённому сетевому туннелю или VPN.

Установка и окружение

Требуются PHP 8.1+ и Localzet Server ^4.3:

composer config repositories.localzet-tunnel vcs https://github.com/localzet/Tunnel
composer require localzet/tunnel:dev-main

Tunnel\Server — обработчик ServerAbstract; создание с socket name вызывает localzet_start() с одним worker. Клиент использует статическое соединение и реестры callbacks внутри процесса.

Подключайте клиента внутри работающего Server/event loop. Выбор среды зависит не только от CLI, но и от того, загружен ли класс Server. Адрес и протокол должны совпадать с listener; для framed-сообщений используют адрес вида frame://127.0.0.1:2206.

Вне Server предусмотрена ветка с stream_socket_client(), однако она пытается присвоить свойство json ресурсу stream. В PHP это ошибка. Не используйте её как рабочий standalone-синхронный клиент без исправления реализации.

События

Внутри обработчика старта своего worker:

use localzet\Tunnel\Client;

Client::connect('frame://127.0.0.1:2206');
Client::on('order.created', static function (array $order): void {
    updateLocalView($order);
});

Client::publish('order.created', ['id' => 42]);

on() связывает канал с callback и отправляет subscribe. Повторный on() для того же имени заменяет локальный callback, а не добавляет второй. Это отличается от localzet\Events, где можно хранить несколько регистраций.

subscribe() принимает имя или массив имён. Он подписывает процесс, но сам не задаёт callback. Если для полученного события нет отдельного callback, клиент вызывает общий Client::$onMessage($channel, $data), если тот задан; иначе выбрасывает исключение.

Имена каналов совпадают точно. Префиксные шаблоны и wildcard-подписки не реализованы.

Broadcast и выдача одному подписчику

publish($channels, $data, false) выбирает серверную команду publish для рассылки всем подписчикам. true выбирает publishLoop: одному подписчику канала по текущему указателю массива.

В broadcast-ветке есть существенная особенность реализации: payload сериализуется в переменную, которая затем повторно сериализуется для следующего получателя. Для нескольких подписчиков последующие сообщения могут содержать закодированную строку вместо ожидаемого объекта envelope. Не рассматривайте одно успешное получение как подтверждение корректной рассылки нескольким клиентам.

Pub/sub не хранит историю для отключённого клиента и не возвращает publisher подтверждение обработки подписчиками.

Очередь заданий

Client::watch('mail.send', static function (array $job): void {
    sendMail($job);
});

Client::enqueue('mail.send', ['message_id' => 42]);

Каждый канал очереди хранит данные в SplQueue. Watcher получает сообщение, когда заявил готовность через reserve(). Из списка готовых consumers он удаляется до отправки задания.

По умолчанию watch(..., $autoReserve = true) запрашивает следующее задание после callback в блоке finally. Это означает, что следующая готовность объявляется и после исключения; reserve не подтверждает успешную обработку прошлого задания.

Для контроля момента следующей выдачи:

Client::watch('mail.send', $handler, false);
Client::reserve();

Теперь приложение само решает, когда снова вызвать reserve(). Повторный вызов относится к очередям, которые наблюдает соединение, а не к отдельному job ID.

Задание удаляется из памяти очереди перед send(). Ack/nack, lease, повторная выдача после падения worker и dead-letter queue не реализованы. Для обязательной обработки с восстановлением после сбоя нужен другой контракт хранения/подтверждений.

Переподключение

onClose() очищает timers, сбрасывает соединение и ставит повторные попытки подключения. onConnect() восстанавливает событийные подписки из реестра events и вызывает пользовательский callback.

Watch-подписки очередей в этой ветке автоматически не восстанавливаются. При необходимости регистрируйте их из Client::$onConnect, учитывая, что старые callbacks уже могут находиться в локальной карте. После reconnect не считайте очередь защищённой от потери ранее выданного задания.

Граница доверия

Сервер и клиент пробуют JSON, затем PHP unserialize() без запрета классов. Не позволяйте недоверенным клиентам отправлять сообщения и не подключайте клиента к недоверенному серверу. Встроенной проверки имени канала, прав подписки, размера сообщения или допустимых команд здесь нет.

Debug-режим пишет содержимое сообщений в Server log. Не включайте его на данных с ключами, bearer-токенами или персональной информацией.

Методы и envelope сообщений

Исходный код · AGPL-3.0