water/server/app/process/MqttSubscriber.php

<?php

namespace app\process;

use app\service\MqttTopic;
use Workerman\Mqtt\Client as MqttClient;
use Webman\Channel\Client as ChannelClient;
use Webman\Event\Event;
use support\Log;

class MqttSubscriber
{
    private ?MqttClient $mqtt = null;

    public function onWorkerStart(): void
    {
        $config = config('mqtt');
        if (empty($config['enabled'])) {
            Log::channel('default')->info('MQTT process is disabled.');
            return;
        }

        ChannelClient::connect();
        ChannelClient::on('mqtt.publish', function (array $message): void {
            $this->publish($message);
        });

        $this->mqtt = new MqttClient($config['address'], $config['client']);
        $this->mqtt->onConnect = function (MqttClient $mqtt) use ($config): void {
            if (empty($config['subscriptions'])) {
                Log::channel('default')->warning('MQTT connected without configured subscriptions.');
                return;
            }

            $mqtt->subscribe($config['subscriptions'], [], function (?\Exception $exception, array $granted): void {
                if ($exception) {
                    Log::channel('default')->error('MQTT subscription failed.', ['exception' => $exception->getMessage()]);
                    return;
                }
                Log::channel('default')->info('MQTT subscriptions are active.', ['granted' => $granted]);
            });
        };
        $this->mqtt->onMessage = function (string $topic, string $payload, MqttClient $mqtt, array $properties = []) use ($config): void {
            if (strlen($payload) > $config['max_payload_bytes']) {
                Log::channel('default')->warning('MQTT message dropped because it exceeds the size limit.', ['topic' => $topic]);
                return;
            }

            $decoded = json_decode($payload, true);
            Log::channel('default')->info('MQTT message received.', ['topic' => $topic, 'bytes' => strlen($payload)]);
            Event::emit('mqtt.message', [
                'topic' => $topic,
                'payload' => $payload,
                'data' => json_last_error() === JSON_ERROR_NONE ? $decoded : null,
                'properties' => $properties,
                'received_at' => date(DATE_ATOM),
            ]);
        };
        $this->mqtt->onError = function (\Exception $exception): void {
            Log::channel('default')->error('MQTT client error.', ['exception' => $exception->getMessage()]);
        };
        $this->mqtt->onClose = function (): void {
            Log::channel('default')->warning('MQTT connection closed; automatic reconnect is active.');
        };
        $this->mqtt->connect();
    }

    private function publish(array $message): void
    {
        if (!$this->mqtt) {
            Log::channel('default')->warning('MQTT publish rejected because the client is unavailable.', ['trace_id' => $message['traceId'] ?? null]);
            return;
        }

        try {
            MqttTopic::assertPublishable((string) ($message['topic'] ?? ''));
            $this->mqtt->publish(
                $message['topic'],
                $message['payload'] ?? '',
                ['qos' => $message['qos'] ?? 1, 'retain' => $message['retain'] ?? false],
                function (?\Exception $exception) use ($message): void {
                    if ($exception) {
                        Log::channel('default')->error('MQTT publish failed.', ['trace_id' => $message['traceId'] ?? null, 'exception' => $exception->getMessage()]);
                    }
                }
            );
        } catch (\Throwable $exception) {
            Log::channel('default')->error('MQTT publish rejected.', ['trace_id' => $message['traceId'] ?? null, 'exception' => $exception->getMessage()]);
        }
    }
}