232 lines
8.1 KiB
PHP
232 lines
8.1 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;
|
||
|
||
/**
|
||
* MQTT 持久化服务：保留消息 + 离线消息队列
|
||
*
|
||
* - 保留消息(retained)：broker 重启后从 DB 恢复，保证新订阅者仍能收到最后一条。
|
||
* - 离线消息(offline)：clean=false 会话离线期间投递给它的 QoS1/2 消息入库，
|
||
* 客户端重连后取出补投。
|
||
*
|
||
* 均为二进制安全（longblob）。所有操作 try/catch，DB 不可用不影响 broker 运行。
|
||
* 由 persist_retain / persist_offline 配置开关控制。
|
||
*/
|
||
class Store
|
||
{
|
||
/** @var bool 启用保留消息持久化 */
|
||
protected $persistRetain;
|
||
|
||
/** @var bool 启用离线消息持久化 */
|
||
protected $persistOffline;
|
||
|
||
/** @var int 单客户端离线消息上限，超出丢弃最旧 */
|
||
protected $offlineLimit;
|
||
|
||
|
||
public function __construct(array $config)
|
||
{
|
||
$this->persistRetain = (string) ($config['persist_retain'] ?? '1') === '1';
|
||
$this->persistOffline = (string) ($config['persist_offline'] ?? '1') === '1';
|
||
$this->offlineLimit = (int) ($config['offline_limit'] ?? 1000);
|
||
}
|
||
|
||
|
||
public function ensureTables(): void
|
||
{
|
||
try {
|
||
Db::execute("CREATE TABLE IF NOT EXISTS `wxapp_mqttbroker_retain` (
|
||
`id` int unsigned NOT NULL AUTO_INCREMENT,
|
||
`topic` varchar(255) NOT NULL DEFAULT '',
|
||
`payload` longblob,
|
||
`qos` tinyint(1) NOT NULL DEFAULT '0',
|
||
`update_at` int DEFAULT NULL,
|
||
PRIMARY KEY (`id`), UNIQUE KEY `uk_topic` (`topic`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
|
||
Db::execute("CREATE TABLE IF NOT EXISTS `wxapp_mqttbroker_offline` (
|
||
`id` int unsigned NOT NULL AUTO_INCREMENT,
|
||
`client_id` varchar(191) NOT NULL DEFAULT '',
|
||
`topic` varchar(255) NOT NULL DEFAULT '',
|
||
`payload` longblob,
|
||
`qos` tinyint(1) NOT NULL DEFAULT '0',
|
||
`create_at` int DEFAULT NULL,
|
||
PRIMARY KEY (`id`), KEY `idx_client_id` (`client_id`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
|
||
Db::execute("CREATE TABLE IF NOT EXISTS `wxapp_mqttbroker_outbox` (
|
||
`id` int unsigned NOT NULL AUTO_INCREMENT,
|
||
`topic` varchar(255) NOT NULL DEFAULT '',
|
||
`payload` longblob,
|
||
`qos` tinyint(1) NOT NULL DEFAULT '0',
|
||
`retain` tinyint(1) NOT NULL DEFAULT '0',
|
||
`source` varchar(32) DEFAULT 'manual',
|
||
`create_at` int DEFAULT NULL,
|
||
`status` tinyint(1) NOT NULL DEFAULT '0' COMMENT '0待投递 1已投递',
|
||
PRIMARY KEY (`id`), KEY `idx_status` (`status`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
|
||
} catch (\Throwable $e) {
|
||
}
|
||
}
|
||
|
||
/* ---------------- 手动发布出站队列 ---------------- */
|
||
|
||
/**
|
||
* 入队一条手动发布（后台/API 发布经 DB 出栈，Broker 消费后内部路由，零外部依赖）
|
||
*/
|
||
public static function enqueueManual(string $topic, string $payload, int $qos, int $retain, string $source): bool
|
||
{
|
||
try {
|
||
Db::name('mqttbroker_outbox')->insert([
|
||
'topic' => $topic,
|
||
'payload' => $payload,
|
||
'qos' => $qos,
|
||
'retain' => $retain ? 1 : 0,
|
||
'source' => $source,
|
||
'create_at' => time(),
|
||
'status' => 0,
|
||
]);
|
||
return true;
|
||
} catch (\Throwable $e) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 取出待投递的出站消息（status=0）
|
||
* @return array
|
||
*/
|
||
public function popOutbox(int $limit = 100): array
|
||
{
|
||
$out = [];
|
||
try {
|
||
$out = Db::name('mqttbroker_outbox')->where('status', 0)
|
||
->order('id', 'asc')->limit($limit)->select()->toArray();
|
||
} catch (\Throwable $e) {
|
||
}
|
||
return $out;
|
||
}
|
||
|
||
/**
|
||
* 标记出站消息已投递（避免重复路由）
|
||
*/
|
||
public function markOutboxSent(array $ids): void
|
||
{
|
||
if (empty($ids)) {
|
||
return;
|
||
}
|
||
try {
|
||
Db::name('mqttbroker_outbox')->whereIn('id', $ids)->update(['status' => 1]);
|
||
} catch (\Throwable $e) {
|
||
}
|
||
}
|
||
|
||
/* ---------------- 保留消息 ---------------- */
|
||
|
||
/**
|
||
* 启动时加载全部保留消息
|
||
* @return array topic => ['payload'=>string,'qos'=>int]
|
||
*/
|
||
public function loadRetained(): array
|
||
{
|
||
if (!$this->persistRetain) {
|
||
return [];
|
||
}
|
||
$out = [];
|
||
try {
|
||
foreach (Db::name('mqttbroker_retain')->select() as $row) {
|
||
$out[$row['topic']] = ['payload' => (string) $row['payload'], 'qos' => (int) $row['qos']];
|
||
}
|
||
} catch (\Throwable $e) {
|
||
}
|
||
return $out;
|
||
}
|
||
|
||
|
||
public function saveRetained(string $topic, string $payload, int $qos): void
|
||
{
|
||
if (!$this->persistRetain) {
|
||
return;
|
||
}
|
||
try {
|
||
$exists = Db::name('mqttbroker_retain')->where('topic', $topic)->find();
|
||
if ($exists) {
|
||
Db::name('mqttbroker_retain')->where('id', $exists['id'])
|
||
->update(['payload' => $payload, 'qos' => $qos, 'update_at' => time()]);
|
||
} else {
|
||
Db::name('mqttbroker_retain')->insert([
|
||
'topic' => $topic, 'payload' => $payload, 'qos' => $qos, 'update_at' => time(),
|
||
]);
|
||
}
|
||
} catch (\Throwable $e) {
|
||
}
|
||
}
|
||
|
||
|
||
public function deleteRetained(string $topic): void
|
||
{
|
||
if (!$this->persistRetain) {
|
||
return;
|
||
}
|
||
try {
|
||
Db::name('mqttbroker_retain')->where('topic', $topic)->delete();
|
||
} catch (\Throwable $e) {
|
||
}
|
||
}
|
||
|
||
/* ---------------- 离线消息 ---------------- */
|
||
|
||
|
||
public function queueOffline(string $clientId, string $topic, string $payload, int $qos): void
|
||
{
|
||
if (!$this->persistOffline) {
|
||
return;
|
||
}
|
||
try {
|
||
Db::name('mqttbroker_offline')->insert([
|
||
'client_id' => $clientId, 'topic' => $topic,
|
||
'payload' => $payload, 'qos' => $qos, 'create_at' => time(),
|
||
]);
|
||
// 超限裁剪：保留最新 offlineLimit 条
|
||
$count = Db::name('mqttbroker_offline')->where('client_id', $clientId)->count();
|
||
if ($count > $this->offlineLimit) {
|
||
$ids = Db::name('mqttbroker_offline')->where('client_id', $clientId)
|
||
->order('id', 'asc')->limit($count - $this->offlineLimit)->column('id');
|
||
if ($ids) {
|
||
Db::name('mqttbroker_offline')->whereIn('id', $ids)->delete();
|
||
}
|
||
}
|
||
} catch (\Throwable $e) {
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 取出并清空某客户端的离线消息
|
||
* @return array<int,array{topic:string,payload:string,qos:int}>
|
||
*/
|
||
public function popOffline(string $clientId): array
|
||
{
|
||
if (!$this->persistOffline) {
|
||
return [];
|
||
}
|
||
$out = [];
|
||
try {
|
||
$rows = Db::name('mqttbroker_offline')->where('client_id', $clientId)->order('id', 'asc')->select();
|
||
foreach ($rows as $r) {
|
||
$out[] = ['topic' => $r['topic'], 'payload' => (string) $r['payload'], 'qos' => (int) $r['qos']];
|
||
}
|
||
if (!empty($out)) {
|
||
Db::name('mqttbroker_offline')->where('client_id', $clientId)->delete();
|
||
}
|
||
} catch (\Throwable $e) {
|
||
}
|
||
return $out;
|
||
}
|
||
}
|