143 lines
4.6 KiB
PHP
143 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 think\facade\Db;
|
||
|
||
/**
|
||
* 规则引擎：匹配源主题过滤器后，将消息转发到外部 Broker（桥接到 EMQX）
|
||
*
|
||
* 规则表 wxapp_mqttbroker_rule：
|
||
* source_filter 源主题过滤器（支持 + / #）
|
||
* target_type mqtt（当前仅支持转发到外部 MQTT Broker）
|
||
* target_broker 外部地址 host:port
|
||
* target_topic 目标主题模板，支持 ${topic} 变量
|
||
* target_qos 转发 QoS
|
||
* enabled 1启用 0停用
|
||
*
|
||
* 行为：
|
||
* - 仅对本地产生的消息生效（distribute 本地路径调用），桥接/外部回流消息不二次转发，避免回环。
|
||
* - $SYS 系统主题不转发。
|
||
* - 规则带 30s TTL 缓存，新增/修改后最多 30s 生效。
|
||
*/
|
||
class RuleEngine
|
||
{
|
||
/** @var array */
|
||
protected $config;
|
||
|
||
/** @var Forwarder */
|
||
protected $forwarder;
|
||
|
||
/** @var array 规则缓存 */
|
||
protected $rules = [];
|
||
|
||
/** @var int 缓存刷新时间戳 */
|
||
protected $loadedAt = 0;
|
||
|
||
/** @var int 缓存有效期（秒） */
|
||
protected $ttl = 30;
|
||
|
||
|
||
public function __construct(array $config, Forwarder $forwarder)
|
||
{
|
||
$this->config = $config;
|
||
$this->forwarder = $forwarder;
|
||
}
|
||
|
||
|
||
public function ensureTables(): void
|
||
{
|
||
try {
|
||
Db::execute("CREATE TABLE IF NOT EXISTS `wxapp_mqttbroker_rule` (
|
||
`id` int unsigned NOT NULL AUTO_INCREMENT,
|
||
`name` varchar(191) NOT NULL DEFAULT '',
|
||
`source_filter` varchar(255) NOT NULL DEFAULT '',
|
||
`target_type` varchar(16) NOT NULL DEFAULT 'mqtt',
|
||
`target_broker` varchar(255) NOT NULL DEFAULT '',
|
||
`target_clientid` varchar(191) DEFAULT '',
|
||
`target_username` varchar(191) DEFAULT '',
|
||
`target_password` varchar(191) DEFAULT '',
|
||
`target_topic` varchar(255) NOT NULL DEFAULT '',
|
||
`target_qos` tinyint(1) NOT NULL DEFAULT '0',
|
||
`enabled` tinyint(1) NOT NULL DEFAULT '1',
|
||
`remark` varchar(255) DEFAULT NULL,
|
||
`create_at` int DEFAULT NULL,
|
||
PRIMARY KEY (`id`), KEY `idx_enabled` (`enabled`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
|
||
} catch (\Throwable $e) {
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 匹配并转发
|
||
*/
|
||
public function matchAndForward(string $topic, string $msg, int $qos): void
|
||
{
|
||
if ((string) ($this->config['rule_enabled'] ?? '1') !== '1') {
|
||
return;
|
||
}
|
||
if (strncmp($topic, '$SYS', 4) === 0) {
|
||
return; // 系统主题不转发
|
||
}
|
||
if (!$this->forwarder->isAvailable()) {
|
||
return;
|
||
}
|
||
foreach ($this->loadRules() as $r) {
|
||
if ((int) $r['enabled'] !== 1 || empty($r['source_filter']) || empty($r['target_broker'])) {
|
||
continue;
|
||
}
|
||
if ($this->topicMatch((string) $r['source_filter'], $topic)) {
|
||
$this->forwarder->forward($r, $topic, $msg, $qos);
|
||
}
|
||
}
|
||
}
|
||
|
||
|
||
protected function loadRules(): array
|
||
{
|
||
$now = time();
|
||
if ($now - $this->loadedAt < $this->ttl && $this->loadedAt > 0) {
|
||
return $this->rules;
|
||
}
|
||
try {
|
||
$this->rules = Db::name('mqttbroker_rule')->where('enabled', 1)
|
||
->order('id', 'asc')->select()->toArray();
|
||
} catch (\Throwable $e) {
|
||
$this->rules = [];
|
||
}
|
||
$this->loadedAt = $now;
|
||
return $this->rules;
|
||
}
|
||
|
||
|
||
protected function topicMatch(string $filter, string $topic): bool
|
||
{
|
||
if ($filter === $topic) {
|
||
return true;
|
||
}
|
||
$f = explode('/', $filter);
|
||
$t = explode('/', $topic);
|
||
foreach ($f as $i => $seg) {
|
||
if ($seg === '#') {
|
||
return true;
|
||
}
|
||
if ($seg === '+') {
|
||
if (!isset($t[$i])) {
|
||
return false;
|
||
}
|
||
continue;
|
||
}
|
||
if (!isset($t[$i]) || $seg !== $t[$i]) {
|
||
return false;
|
||
}
|
||
}
|
||
return count($f) === count($t);
|
||
}
|
||
}
|