124 lines
4.6 KiB
PHP
124 lines
4.6 KiB
PHP
<?php
|
||
// +----------------------------------------------------------------------
|
||
// | YwxApp [ WE CAN DO IT JUST THINK ]
|
||
// +----------------------------------------------------------------------
|
||
// | Copyright (c) 2026-2036 http://ywxapp.cn All rights reserved.
|
||
// +----------------------------------------------------------------------
|
||
// | Author: ywxapp<admin@ywxapp.cn>
|
||
// +----------------------------------------------------------------------
|
||
namespace addon\mqttbroker\service;
|
||
|
||
use Workerman\Mqtt\Client;
|
||
|
||
/**
|
||
* 外部 Broker 转发器（规则引擎的投递通道）
|
||
*
|
||
* 将本地匹配到的消息转发到外部 MQTT Broker（如 EMQX），实现"桥接到 EMQX"。
|
||
* 每个目标 Broker 维护一个异步常驻连接（基于 Workerman\Mqtt\Client，复用 Broker 的事件循环），
|
||
* 连接未就绪时消息进入待发缓冲（上限 200，避免无限堆积），连接成功后冲刷。
|
||
*
|
||
* 依赖 workerman/mqtt（composer require workerman/mqtt）。未安装时 isAvailable()=false，
|
||
* 规则引擎自动跳过转发，不影响 Broker 其它功能。
|
||
*/
|
||
class Forwarder
|
||
{
|
||
/** @var array key => ['client'=>Client|null,'connected'=>bool,'pending'=>callable[]] */
|
||
protected $clients = [];
|
||
|
||
|
||
public function isAvailable(): bool
|
||
{
|
||
return class_exists(Client::class);
|
||
}
|
||
|
||
/**
|
||
* 按规则转发一条消息到外部 Broker
|
||
*/
|
||
public function forward(array $rule, string $topic, string $msg, int $qos): void
|
||
{
|
||
if (!$this->isAvailable()) {
|
||
return;
|
||
}
|
||
$key = md5(($rule['target_broker'] ?? '') . '|' . ($rule['target_clientid'] ?? '') . '|' . ($rule['target_username'] ?? ''));
|
||
if (!isset($this->clients[$key]) || $this->clients[$key]['client'] === null) {
|
||
$this->connect($key, $rule);
|
||
}
|
||
$entry = &$this->clients[$key];
|
||
$target = $this->buildTargetTopic((string) ($rule['target_topic'] ?? ''), $topic);
|
||
$pubQos = min($qos, (int) ($rule['target_qos'] ?? 0));
|
||
|
||
$send = function () use ($entry, $target, $msg, $pubQos) {
|
||
try {
|
||
$entry['client']->publish($target, $msg, ['qos' => $pubQos]);
|
||
} catch (\Throwable $e) {
|
||
}
|
||
};
|
||
|
||
if ($entry['connected']) {
|
||
$send();
|
||
} elseif (count($entry['pending']) < 200) {
|
||
$entry['pending'][] = $send;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 建立到目标 Broker 的异步连接（懒连接）
|
||
*/
|
||
protected function connect(string $key, array $rule): void
|
||
{
|
||
$this->clients[$key] = ['client' => null, 'connected' => false, 'pending' => []];
|
||
try {
|
||
$uri = $this->buildUri((string) ($rule['target_broker'] ?? ''));
|
||
$opts = [];
|
||
if (!empty($rule['target_clientid'])) {
|
||
$opts['clientId'] = $rule['target_clientid'];
|
||
}
|
||
if (!empty($rule['target_username'])) {
|
||
$opts['username'] = $rule['target_username'];
|
||
$opts['password'] = $rule['target_password'] ?? '';
|
||
}
|
||
$client = new Client($uri, $opts);
|
||
$self = $this;
|
||
$client->onConnect = function () use ($key, $self) {
|
||
$self->clients[$key]['connected'] = true;
|
||
foreach ($self->clients[$key]['pending'] as $cb) {
|
||
try { $cb(); } catch (\Throwable $e) {
|
||
}
|
||
}
|
||
$self->clients[$key]['pending'] = [];
|
||
};
|
||
$client->onError = function () use ($key, $self) {
|
||
$self->clients[$key]['connected'] = false;
|
||
};
|
||
$client->onClose = function () use ($key, $self) {
|
||
$self->clients[$key]['connected'] = false;
|
||
};
|
||
$client->connect();
|
||
$this->clients[$key]['client'] = $client;
|
||
} catch (\Throwable $e) {
|
||
$this->clients[$key] = ['client' => null, 'connected' => false, 'pending' => []];
|
||
}
|
||
}
|
||
|
||
|
||
protected function buildUri(string $broker): string
|
||
{
|
||
$broker = trim($broker);
|
||
if ($broker === '' || strpos($broker, '://') !== false) {
|
||
return $broker === '' ? 'mqtt://127.0.0.1:1883' : $broker;
|
||
}
|
||
return 'mqtt://' . $broker;
|
||
}
|
||
|
||
/**
|
||
* 目标主题模板：空或 ${topic} 表示沿用原主题，否则替换变量
|
||
*/
|
||
protected function buildTargetTopic(string $tpl, string $srcTopic): string
|
||
{
|
||
if ($tpl === '' || $tpl === '${topic}') {
|
||
return $srcTopic;
|
||
}
|
||
return str_replace('${topic}', $srcTopic, $tpl);
|
||
}
|
||
}
|