refactor(core): 拆分 model/mqtt/update 为单一职责子模块

- model/ 按领域拆为 connection/subscription/message/publish/variable/log/settings
- mqtt/ 拆为门面(引擎与事件)、tls、v5、v4 四块
- update/ 拆为门面(版本工具与命名约定)、check(检查)、install(下载/安装/清理)
- sysmon 只保留进程资源采样,窗口置顶迁至新 platform.rs
- 对外路径经 mod.rs 再导出保持不变,测试随门面保留(33 通过)
This commit is contained in:
2026-09-29 22:20:53 +08:00
parent 303b348ffe
commit b5089f5898
21 changed files with 3393 additions and 3252 deletions
+1
View File
@@ -3,6 +3,7 @@
pub mod aliyun; pub mod aliyun;
pub mod model; pub mod model;
pub mod mqtt; pub mod mqtt;
pub mod platform;
pub mod store; pub mod store;
pub mod sysmon; pub mod sysmon;
pub mod ui; pub mod ui;
File diff suppressed because it is too large Load Diff
+277
View File
@@ -0,0 +1,277 @@
//! 连接配置:协议版本、传输方式、遗嘱、TLS 与连接参数。
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
#[derive(Default)]
pub enum ProtocolVersion {
/// MQTT 3.1.1
V311,
/// MQTT 5.0
#[default]
V5,
}
impl ProtocolVersion {
pub const ALL: [Self; 2] = [Self::V5, Self::V311];
pub fn label(self) -> &'static str {
match self {
Self::V311 => "MQTT 3.1.1",
Self::V5 => "MQTT 5.0",
}
}
pub fn is_v5(self) -> bool {
matches!(self, Self::V5)
}
}
/// 传输方式。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[derive(Default)]
pub enum TransportKind {
/// 明文 TCP,默认 1883
#[default]
Tcp,
/// TLS over TCP,默认 8883
Tls,
/// WebSocket (ws://),默认 8083
Ws,
/// WebSocket over TLS (wss://),默认 8084
Wss,
}
impl TransportKind {
pub const ALL: [Self; 4] = [Self::Tcp, Self::Tls, Self::Ws, Self::Wss];
pub fn label(self) -> &'static str {
match self {
Self::Tcp => "TCP",
Self::Tls => "TLS",
Self::Ws => "WebSocket",
Self::Wss => "WSS",
}
}
pub fn default_port(self) -> u16 {
match self {
Self::Tcp => 1883,
Self::Tls => 8883,
Self::Ws => 8083,
Self::Wss => 8084,
}
}
pub fn is_tls(self) -> bool {
matches!(self, Self::Tls | Self::Wss)
}
pub fn is_websocket(self) -> bool {
matches!(self, Self::Ws | Self::Wss)
}
}
/// MQTT 5 遗嘱消息(3.1.1 下忽略 v5 专属属性)。
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct LastWill {
pub topic: String,
pub payload: String,
pub qos: u8,
pub retain: bool,
#[serde(default)]
pub content_type: Option<String>,
#[serde(default)]
pub response_topic: Option<String>,
}
/// SSL/TLS 自定义配置(CA、双向认证、跳过校验)。
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct SslConfig {
/// 自定义 CA 文件路径(PEM),作为系统根证书的补充
#[serde(default)]
pub ca_file: String,
/// 客户端证书(PEM),双向认证用
#[serde(default)]
pub client_cert_file: String,
/// 客户端私钥(PEM),与客户端证书配套
#[serde(default)]
pub client_key_file: String,
/// 跳过服务器证书校验(insecure),仅调试环境使用
#[serde(default)]
pub ignore_ca: bool,
}
impl SslConfig {
/// 是否完全未配置(走 rumqttc 默认 TLS 即可)。
pub fn is_default(&self) -> bool {
!self.ignore_ca
&& self.ca_file.is_empty()
&& self.client_cert_file.is_empty()
&& self.client_key_file.is_empty()
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConnectionConfig {
pub id: String,
pub name: String,
/// 主机(不含端口、协议前缀)
pub host: String,
pub port: u16,
/// WebSocket 路径,如 /mqtt
#[serde(default)]
pub path: String,
#[serde(default)]
pub protocol: ProtocolVersion,
#[serde(default)]
pub transport: TransportKind,
pub client_id: String,
#[serde(default)]
pub username: String,
#[serde(default)]
pub password: String,
#[serde(default = "default_true")]
pub clean_start: bool,
#[serde(default = "default_keep_alive")]
pub keep_alive: u16,
// ── MQTT 5 专属 ──
#[serde(default)]
pub session_expiry_interval: u32,
#[serde(default)]
pub receive_maximum: Option<u16>,
#[serde(default)]
pub maximum_packet_size: Option<u32>,
#[serde(default)]
pub topic_alias_maximum: Option<u16>,
/// 连接成功后自动恢复已保存的订阅
#[serde(default = "default_true")]
pub auto_resubscribe: bool,
/// 断线自动重连
#[serde(default = "default_true")]
pub auto_reconnect: bool,
/// 应用启动时自动连接该连接
#[serde(default)]
pub auto_connect: bool,
#[serde(default)]
pub last_will: Option<LastWill>,
/// 创建时间(unix 秒),用于排序
#[serde(default)]
pub created_at: i64,
/// 连接分组(对标 MQTTX folder)
#[serde(default)]
pub group: Option<String>,
/// 连接超时(秒),覆盖建连与 CONNACK 等待
#[serde(default = "default_connection_timeout")]
pub connection_timeout_secs: u16,
/// 最大重连次数,0 = 无限重连
#[serde(default)]
pub max_reconnect_times: u32,
/// SSL/TLS 自定义配置
#[serde(default)]
pub ssl: SslConfig,
}
pub(super) fn default_true() -> bool {
true
}
pub(super) fn default_keep_alive() -> u16 {
60
}
pub(super) fn default_connection_timeout() -> u16 {
10
}
/// 生成随机 Client ID(新建连接与表单「重新生成」共用同一逻辑)。
pub fn generate_client_id() -> String {
format!("mqttx_{}", &uuid::Uuid::new_v4().simple().to_string()[..8])
}
impl ConnectionConfig {
pub fn new() -> Self {
let transport = TransportKind::default();
Self {
id: uuid::Uuid::new_v4().to_string(),
name: "新建连接".into(),
host: "broker.emqx.io".into(),
port: transport.default_port(),
path: "/mqtt".into(),
protocol: ProtocolVersion::V5,
transport,
client_id: generate_client_id(),
username: String::new(),
password: String::new(),
clean_start: true,
keep_alive: 60,
session_expiry_interval: 0,
receive_maximum: None,
maximum_packet_size: None,
topic_alias_maximum: None,
auto_resubscribe: true,
auto_reconnect: true,
auto_connect: false,
last_will: None,
created_at: chrono::Local::now().timestamp(),
group: None,
connection_timeout_secs: default_connection_timeout(),
max_reconnect_times: 0,
ssl: SslConfig::default(),
}
}
/// broker.emqx.io 等公共服务器的快速预设。
pub fn with_public(name: &str, host: &str, port: u16, transport: TransportKind) -> Self {
let mut c = Self::new();
c.name = name.into();
c.host = host.into();
c.port = port;
c.transport = transport;
c
}
pub fn display_address(&self) -> String {
format!("{}:{}", self.host, self.port)
}
/// 与另一份配置相比,影响 broker 会话建立的关键参数是否变化。
/// 这些字段改动只有在重新建立连接后才会生效,用于提示用户重连。
/// WS 路径、遗嘱、TLS 配置、keep_alive 同样随 CONNECT 报文一次性生效,
/// 必须纳入比较。
pub fn session_params_changed(&self, other: &ConnectionConfig) -> bool {
self.host != other.host
|| self.port != other.port
|| self.path != other.path
|| self.transport != other.transport
|| self.protocol != other.protocol
|| self.client_id != other.client_id
|| self.username != other.username
|| self.password != other.password
|| self.clean_start != other.clean_start
|| self.keep_alive != other.keep_alive
|| self.last_will != other.last_will
|| self.ssl != other.ssl
}
}
impl Default for ConnectionConfig {
fn default() -> Self {
Self::new()
}
}
// ─── 订阅 ────────────────────────────────────────────────────────────────────
+45
View File
@@ -0,0 +1,45 @@
//! 运行日志:日志级别与日志条目。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LogLevel {
Debug,
Info,
Warn,
Error,
}
impl LogLevel {
pub fn label(self) -> &'static str {
match self {
Self::Debug => "DEBUG",
Self::Info => "INFO",
Self::Warn => "WARN",
Self::Error => "ERROR",
}
}
pub fn parse(s: &str) -> Self {
match s {
"debug" => Self::Debug,
"warn" => Self::Warn,
"error" => Self::Error,
_ => Self::Info,
}
}
}
#[derive(Debug, Clone)]
pub struct LogEntry {
pub timestamp: i64,
pub connection_id: String,
pub level: LogLevel,
pub event: String,
pub message: String,
pub details: Option<String>,
}
// ─── 连接状态 ────────────────────────────────────────────────────────────────
+199
View File
@@ -0,0 +1,199 @@
//! 消息:方向、载荷格式、消息记录与条数/字节双预算的环形缓冲。
use std::sync::Arc;
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Direction {
Received,
Published,
}
impl Direction {
pub fn label(self) -> &'static str {
match self {
Self::Received => "接收",
Self::Published => "发布",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[derive(Default)]
pub enum PayloadFormat {
#[default]
Plaintext,
Json,
Base64,
Hex,
}
impl PayloadFormat {
pub const ALL: [Self; 4] = [Self::Plaintext, Self::Json, Self::Base64, Self::Hex];
pub fn label(self) -> &'static str {
match self {
Self::Plaintext => "Plaintext",
Self::Json => "JSON",
Self::Base64 => "Base64",
Self::Hex => "Hex",
}
}
pub fn encode(self, text: &str) -> Result<Vec<u8>, String> {
match self {
Self::Plaintext | Self::Json => Ok(text.as_bytes().to_vec()),
Self::Base64 => {
use base64::Engine as _;
base64::engine::general_purpose::STANDARD
.decode(text.trim())
.map_err(|e| format!("Base64 解码失败: {e}"))
}
Self::Hex => {
let cleaned: String = text.chars().filter(|c| !c.is_whitespace()).collect();
if !cleaned.len().is_multiple_of(2) {
return Err("Hex 长度必须为偶数(每两个字符一个字节)".into());
}
// 必须先做全字节校验:长度检查按字节计数,多字节字符(如中文)
// 可凑成偶数字节,直接按字节索引切片会落在字符边界内而 panic。
if !cleaned.bytes().all(|b| b.is_ascii_hexdigit()) {
return Err("Hex 解码失败: 含非十六进制字符".into());
}
(0..cleaned.len())
.step_by(2)
.map(|i| {
u8::from_str_radix(&cleaned[i..i + 2], 16)
.map_err(|e| format!("Hex 解码失败: {e}"))
})
.collect()
}
}
}
}
/// 一条收到/发出的消息记录。payload 以 UTF-8 文本保存(非文本按 lossy 显示)。
///
/// 内存优化:`connection_id`/`topic` 用 `Arc<str>`——同一连接内主题高度重复,
/// 驻留后数千条记录只保留一份字符串。
#[derive(Debug, Clone)]
pub struct MqttRecord {
pub seq: u64,
pub connection_id: Arc<str>,
pub direction: Direction,
pub topic: Arc<str>,
/// Arc 共享:消息列表渲染每帧克隆最多数百条记录,零拷贝
pub payload: Arc<str>,
pub qos: u8,
pub retain: bool,
/// unix 毫秒
pub timestamp: i64,
pub user_properties: Vec<(String, String)>,
pub content_type: Option<String>,
pub response_topic: Option<String>,
pub correlation_data: Option<String>,
pub message_expiry_interval: Option<u32>,
pub subscription_identifier: Option<u32>,
/// payload 超过 [`MAX_PAYLOAD_RETAIN`] 被截断
pub payload_truncated: bool,
/// 非文本负载的原始字节(合法 UTF-8 时为 None)。文本化会失真
/// (非法字节被替换为 U+FFFD),详情面板的 Hex/Base64 需要真实报文。
pub raw_bytes: Option<Arc<[u8]>>,
}
impl MqttRecord {
/// 该记录保留的主要字节数(文本 + 原始字节),用于内存预算驱逐。
/// 其余字段(主题/属性等)相对小一个数量级,不计入。
pub fn retained_bytes(&self) -> usize {
self.payload.len() + self.raw_bytes.as_ref().map_or(0, |b| b.len())
}
}
/// 单连接的消息环形缓冲:条数上限 + [`MAX_CONNECTION_MESSAGE_BYTES`]
/// 字节上限双重驱逐(从最旧开始,均摊 O(1))。
pub struct MessageRing {
queue: VecDeque<MqttRecord>,
bytes: usize,
}
impl MessageRing {
pub fn new(max_count: usize) -> Self {
Self {
// 预分配按较小值钳制:设置上限十万条时不一次性分配过大
queue: VecDeque::with_capacity(max_count.clamp(16, 512)),
bytes: 0,
}
}
/// `max_count` 每次传入以便运行时调整设置后即时生效。
/// 设置层已校验 ≥100,这里只防 0 值导致缓冲被清空。
pub fn push(&mut self, record: MqttRecord, max_count: usize) {
let max_count = max_count.max(1);
self.bytes += record.retained_bytes();
while self.queue.len() >= max_count
|| (self.bytes > MAX_CONNECTION_MESSAGE_BYTES && self.queue.len() > 1)
{
let Some(old) = self.queue.pop_front() else { break };
self.bytes -= old.retained_bytes();
}
self.queue.push_back(record);
}
pub fn len(&self) -> usize {
self.queue.len()
}
pub fn is_empty(&self) -> bool {
self.queue.is_empty()
}
/// 当前保留的字节数(payload + 原始字节)。
pub fn retained_bytes(&self) -> usize {
self.bytes
}
pub fn iter(&self) -> std::collections::vec_deque::Iter<'_, MqttRecord> {
self.queue.iter()
}
}
/// 消息历史中单条 payload 的保留上限。超限截断,避免个别大报文
/// 长期占据历史缓冲(MQTTX 亦有类似限制)。
pub const MAX_PAYLOAD_RETAIN: usize = 128 * 1024;
/// 每连接消息历史的字节预算(payload + 原始字节)。条数上限
/// (max_messages)在大报文流下形同虚设(2000 条 × 128KB ≈ 256MB),
/// 超出预算后从最旧开始驱逐,保证单连接消息内存有上界。
pub const MAX_CONNECTION_MESSAGE_BYTES: usize = 32 * 1024 * 1024;
/// 把原始 payload 字节转成用于历史记录的共享文本,超限截断。
/// 返回 (文本, 非文本时的原始字节, 是否截断)。截断时回退到 UTF-8 字符边界。
///
/// 仅当文本化会失真(非合法 UTF-8,字节被替换为 U+FFFD)时才保留原始
/// 字节(同样受 [`MAX_PAYLOAD_RETAIN`] 限制);纯文本负载零额外开销。
pub fn retained_payload(raw: &[u8]) -> (Arc<str>, Option<Arc<[u8]>>, bool) {
let truncated = raw.len() > MAX_PAYLOAD_RETAIN;
let text_part = if truncated {
let mut end = MAX_PAYLOAD_RETAIN;
while end > 0 && (raw[end] & 0xC0) == 0x80 {
end -= 1;
}
&raw[..end]
} else {
raw
};
let raw_bytes = match std::str::from_utf8(text_part) {
Ok(_) => None,
Err(_) => Some(Arc::from(text_part.to_vec())),
};
let text: Arc<str> = Arc::from(String::from_utf8_lossy(text_part).as_ref());
(text, raw_bytes, truncated)
}
// ─── 发布参数 / 发布预设 ──────────────────────────────────────────────────────
+379
View File
@@ -0,0 +1,379 @@
//! 应用数据模型:按领域拆分的纯数据层(不依赖 GPUI)。
//!
//! - [`connection`] 连接配置与协议/传输枚举
//! - [`subscription`] 订阅与订阅选项
//! - [`message`] 消息记录与环形缓冲
//! - [`publish`] 发布参数与发布预设
//! - [`variable`] 全局变量与模板渲染
//! - [`log`] 运行日志
//! - [`settings`] 应用设置与主题偏好
mod connection;
mod log;
mod message;
mod publish;
mod settings;
mod subscription;
mod variable;
pub use connection::*;
pub use log::*;
pub use message::*;
pub use publish::*;
pub use settings::*;
pub use subscription::*;
pub use variable::*;
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
// 修复回归:Hex 奇数长度必须报错,不能静默截断
#[test]
fn hex_odd_length_is_rejected() {
let err = PayloadFormat::Hex.encode("abc").unwrap_err();
assert!(err.contains("偶数"), "应提示偶数长度,实际: {err}");
assert!(PayloadFormat::Hex.encode("").is_ok());
assert_eq!(
PayloadFormat::Hex.encode("68 69").unwrap(),
vec![0x68, 0x69],
"空白应被忽略"
);
assert!(PayloadFormat::Hex.encode("zz").is_err());
}
#[test]
fn base64_encode_roundtrip() {
use base64::Engine as _;
let text = "hello 世界";
let encoded = base64::engine::general_purpose::STANDARD.encode(text);
assert_eq!(PayloadFormat::Base64.encode(&encoded).unwrap(), text.as_bytes());
assert!(PayloadFormat::Base64.encode("not base64!!!").is_err());
}
#[test]
fn plaintext_and_json_encode_as_utf8() {
assert_eq!(
PayloadFormat::Plaintext.encode("中文").unwrap(),
"中文".as_bytes()
);
assert_eq!(
PayloadFormat::Json.encode(r#"{"a":1}"#).unwrap(),
r#"{"a":1}"#.as_bytes()
);
}
// 修复回归:raw_bytes 是运行时字段,不得写入持久化 JSON
#[test]
fn raw_bytes_not_serialized() {
let p = PublishParams {
topic: "t".into(),
payload: "hi".into(),
raw_bytes: Some(vec![0x00, 0xff]),
..Default::default()
};
let json = serde_json::to_string(&p).unwrap();
assert!(!json.contains("raw_bytes"), "raw_bytes 不应被序列化: {json}");
let back: PublishParams = serde_json::from_str(&json).unwrap();
assert!(back.raw_bytes.is_none());
// 反序列化时输入中的 raw_bytes 字段应被忽略(skip 双向生效)
let mut with_junk = serde_json::to_string(&PublishParams::default()).unwrap();
with_junk.pop(); // 去掉结尾 '}'
with_junk.push_str(r#", "raw_bytes":[1,2]}"#);
let back: PublishParams = serde_json::from_str(&with_junk).unwrap();
assert!(back.raw_bytes.is_none(), "#[serde(skip)] 应忽略输入字段");
}
// 修复回归:同一条消息内 $ts 与 $ts_ms 基于同一时刻
#[test]
fn template_timestamps_share_one_timebase() {
let out = render_template("s={{$ts}} ms={{$ts_ms}}", &[]);
let (s_part, ms_part) = out.split_once(" ms=").expect("格式 s=.. ms=..");
let s: i64 = s_part
.strip_prefix("s=")
.expect("秒字段")
.parse()
.unwrap();
let ms: i64 = ms_part.parse().expect("毫秒字段");
assert_eq!(s, ms / 1000, "秒与毫秒应来自同一时间基准: {out}");
}
#[test]
fn template_unknown_placeholder_kept_and_custom_var_resolved() {
assert_eq!(render_template("{{$unknown}}", &[]), "{{$unknown}}");
let vars = vec![GlobalVariable {
key: "device".into(),
value: "sensor-1".into(),
}];
assert!(render_template("d={{device}}/{{$uuid}}", &vars).starts_with("d=sensor-1/"));
// 多字节字符不被拆坏
assert_eq!(render_template("中文{{$no}}", &[]), "中文{{$no}}");
}
// 连接前遗嘱模板注入:只改副本、空遗嘱不 panic、v5 属性一并渲染
#[test]
fn render_will_templates_injects_and_keeps_no_will_safe() {
let vars = vec![GlobalVariable {
key: "device".into(),
value: "sensor-9".into(),
}];
// 无遗嘱:直接返回
let mut plain = ConnectionConfig::new();
render_will_templates(&mut plain, &vars);
assert!(plain.last_will.is_none());
let mut cfg = ConnectionConfig::new();
cfg.last_will = Some(LastWill {
topic: "alarm/{{device}}".into(),
payload: "d={{device}}".into(),
qos: 1,
retain: false,
content_type: Some("text/{{device}}".into()),
response_topic: Some("ack/{{device}}".into()),
});
render_will_templates(&mut cfg, &vars);
let w = cfg.last_will.as_ref().unwrap();
assert_eq!(w.topic, "alarm/sensor-9");
assert_eq!(w.payload, "d=sensor-9");
assert_eq!(w.content_type.as_deref(), Some("text/sensor-9"));
assert_eq!(w.response_topic.as_deref(), Some("ack/sensor-9"));
// 原始模板串不因注入而需要回写:连接配置持久化的是模板原文,
// 这里只验证副本注入结果;未知占位符保留
let mut raw = ConnectionConfig::new();
raw.last_will = Some(LastWill {
topic: "t/{{unknown}}".into(),
..Default::default()
});
render_will_templates(&mut raw, &vars);
assert_eq!(raw.last_will.as_ref().unwrap().topic, "t/{{unknown}}");
}
// 预设持久化往返:raw_bytes 不落盘,参数完整保留
#[test]
fn publish_preset_roundtrip_skips_raw_bytes() {
let preset = PublishPreset {
id: "p1".into(),
name: "报警".into(),
connection_id: Some("c1".into()),
params: PublishParams {
topic: "alarm/{{device}}".into(),
payload: "6869".into(),
payload_format: PayloadFormat::Hex,
qos: 1,
retain: true,
content_type: Some("application/json".into()),
user_properties: vec![("k".into(), "v".into())],
raw_bytes: Some(vec![0x68, 0x69]),
..Default::default()
},
};
let json = serde_json::to_string(&preset).unwrap();
let back: PublishPreset = serde_json::from_str(&json).unwrap();
assert_eq!(back.name, "报警");
assert_eq!(back.params.topic, "alarm/{{device}}");
assert_eq!(back.params.payload_format, PayloadFormat::Hex);
assert_eq!(back.params.qos, 1);
assert!(back.params.retain);
assert_eq!(back.params.content_type.as_deref(), Some("application/json"));
assert_eq!(back.params.user_properties.len(), 1);
assert!(back.params.raw_bytes.is_none());
}
#[test]
fn payload_format_equality_and_default() {
assert_eq!(PayloadFormat::default(), PayloadFormat::Plaintext);
assert_ne!(PayloadFormat::Hex, PayloadFormat::Base64);
let json = serde_json::to_string(&PayloadFormat::Hex).unwrap();
assert_eq!(
serde_json::from_str::<PayloadFormat>(&json).unwrap(),
PayloadFormat::Hex
);
}
// 向后兼容:旧版 connections.json 不含 Phase 1 新字段,必须能反序列化且取默认值
#[test]
fn legacy_connection_json_fills_new_defaults() {
let legacy = r#"{
"id": "c1",
"name": "旧连接",
"host": "broker.emqx.io",
"port": 1883,
"client_id": "legacy_cid",
"keep_alive": 30
}"#;
let c: ConnectionConfig = serde_json::from_str(legacy).expect("旧版 JSON 应能反序列化");
assert_eq!(c.name, "旧连接");
assert_eq!(c.keep_alive, 30);
assert!(c.group.is_none(), "group 默认应为 None");
assert_eq!(c.connection_timeout_secs, 10, "连接超时默认 10 秒");
assert_eq!(c.max_reconnect_times, 0, "最大重连次数默认 0(无限)");
assert!(c.ssl.ca_file.is_empty());
assert!(c.ssl.client_cert_file.is_empty());
assert!(c.ssl.client_key_file.is_empty());
assert!(!c.ssl.ignore_ca, "ignore_ca 默认关闭");
}
// 向后兼容:旧版订阅 JSON 必须默认 enabled=true,其余 v5 选项为关闭/None
#[test]
fn legacy_subscription_json_defaults_enabled() {
let legacy = r#"{"id":"s1","connection_id":"c1","topic":"a/b","qos":1}"#;
let s: Subscription = serde_json::from_str(legacy).expect("旧版订阅应能反序列化");
assert!(s.enabled, "旧订阅默认启用");
assert!(s.alias.is_none());
assert!(s.sub_identifier.is_none());
assert!(!s.no_local);
assert!(!s.retain_as_published);
assert_eq!(s.retain_handling, 0);
// Subscription::new 同样产出默认启用的订阅
assert!(Subscription::new("c1", "t", 0).enabled);
}
// payload 截断:上限内不截断;超出时截断且不切坏 UTF-8 字符
#[test]
fn retained_payload_caps_and_preserves_utf8() {
let small = b"hello";
let (p, raw, trunc) = retained_payload(small);
assert_eq!(&*p, "hello");
assert!(raw.is_none(), "合法 UTF-8 不保留原始字节");
assert!(!trunc);
// 中文每字符 3 字节,让上限落在字符中间(256K 不是 3 的倍数)
let big: Vec<u8> = "中".repeat(MAX_PAYLOAD_RETAIN).into_bytes();
let (p, _, trunc) = retained_payload(&big);
assert!(trunc, "超过上限必须截断");
assert!(p.len() <= MAX_PAYLOAD_RETAIN);
assert_eq!(p.chars().last(), Some('中'), "不能切坏 UTF-8 字符");
assert!(
big.starts_with(p.as_bytes()),
"截断内容必须是原数据前缀"
);
// 恰好上限不截断
let exact = vec![b'a'; MAX_PAYLOAD_RETAIN];
let (_, raw, trunc) = retained_payload(&exact);
assert!(!trunc);
assert!(raw.is_none());
}
// 非文本负载:保留原始字节供 Hex/Base64 详情,文本侧 lossy 展示
#[test]
fn retained_payload_keeps_raw_bytes_for_binary() {
let binary = vec![0x00, 0xff, 0xfe, 0x01];
let (p, raw, trunc) = retained_payload(&binary);
assert!(!trunc);
assert_eq!(&*p, String::from_utf8_lossy(&binary));
let raw = raw.expect("非法 UTF-8 必须保留原始字节");
assert_eq!(&*raw, &binary);
}
// Hex 编码:多字节字符不得 panic(曾按字节切片落入字符边界内而崩溃)
#[test]
fn hex_encode_rejects_non_hex_without_panic() {
assert!(PayloadFormat::Hex.encode("deadBEEF").is_ok());
assert!(PayloadFormat::Hex.encode("de ad be ef").is_ok());
assert!(PayloadFormat::Hex.encode("abc").is_err(), "奇数长度报错");
assert!(PayloadFormat::Hex.encode("zz").is_err(), "非法字符报错");
// 偶数字节长度的多字节字符:修复前此处在字符边界内切片 panic
assert!(PayloadFormat::Hex.encode("中文").is_err());
assert!(PayloadFormat::Hex.encode("😀").is_err());
}
// 会话参数比较必须覆盖所有随 CONNECT 报文一次性生效的字段
#[test]
fn session_params_changed_covers_connect_time_fields() {
let a = ConnectionConfig::default();
let mut b = a.clone();
assert!(!a.session_params_changed(&b));
b.keep_alive = a.keep_alive + 1;
assert!(a.session_params_changed(&b), "keep_alive 改动应触发重连提示");
b = a.clone();
b.last_will = Some(LastWill {
topic: "w/t".into(),
..Default::default()
});
assert!(a.session_params_changed(&b), "遗嘱改动应触发重连提示");
b = a.clone();
b.ssl.ca_file = "/tmp/ca.pem".into();
assert!(a.session_params_changed(&b), "TLS 配置改动应触发重连提示");
}
fn ring_record(seq: u64, payload_len: usize, raw_len: usize) -> MqttRecord {
MqttRecord {
seq,
connection_id: "c".into(),
direction: Direction::Received,
topic: "t".into(),
payload: Arc::from("a".repeat(payload_len)),
qos: 0,
retain: false,
timestamp: 0,
user_properties: Vec::new(),
content_type: None,
response_topic: None,
correlation_data: None,
message_expiry_interval: None,
subscription_identifier: None,
payload_truncated: false,
raw_bytes: if raw_len > 0 {
Some(Arc::from(vec![0u8; raw_len]))
} else {
None
},
}
}
// 消息环形缓冲:条数上限正常驱逐
#[test]
fn message_ring_enforces_count_limit() {
let mut ring = MessageRing::new(4);
for i in 0..10 {
ring.push(ring_record(i, 64, 0), 4);
}
assert_eq!(ring.len(), 4);
assert_eq!(
ring.iter().next().map(|r| r.seq),
Some(6),
"应从最旧开始驱逐"
);
assert_eq!(ring.retained_bytes(), 4 * 64);
}
// 消息环形缓冲:大报文流下字节预算优先于条数上限,内存有上界
#[test]
fn message_ring_enforces_byte_budget() {
let mut ring = MessageRing::new(2000);
// 每条 256KB(payload 128KB + raw 128KB):2000 条本会到 512MB
let per = MAX_PAYLOAD_RETAIN;
for i in 0..2000 {
ring.push(ring_record(i, per, per), 2000);
}
assert!(
ring.retained_bytes() <= MAX_CONNECTION_MESSAGE_BYTES + 2 * per,
"字节数 {} 超出预算 {}",
ring.retained_bytes(),
MAX_CONNECTION_MESSAGE_BYTES
);
assert!(
ring.len() < 2000,
"大报文流不应保留满额条数(实际 {} 条)",
ring.len()
);
assert_eq!(ring.iter().next().map(|r| r.seq), Some((2000 - ring.len()) as u64));
}
// 单条超预算的记录也至少保留最新一条(列表不能被清空)
#[test]
fn message_ring_never_empties_on_oversized_record() {
let mut ring = MessageRing::new(10);
ring.push(ring_record(0, MAX_PAYLOAD_RETAIN, MAX_PAYLOAD_RETAIN), 10);
assert_eq!(ring.len(), 1);
ring.push(ring_record(1, 64, 0), 10);
assert_eq!(ring.len(), 2);
assert_eq!(ring.iter().next().map(|r| r.seq), Some(0));
}
}
+57
View File
@@ -0,0 +1,57 @@
//! 发布:发布参数与可保存的发布预设。
use serde::{Deserialize, Serialize};
use super::*;
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct PublishParams {
pub topic: String,
/// 用于界面展示与记录的文本负载。
pub payload: String,
#[serde(default)]
pub payload_format: PayloadFormat,
pub qos: u8,
pub retain: bool,
/// MQTT 5 用户属性
#[serde(default)]
pub user_properties: Vec<(String, String)>,
#[serde(default)]
pub content_type: Option<String>,
#[serde(default)]
pub message_expiry_interval: Option<u32>,
#[serde(default)]
pub response_topic: Option<String>,
#[serde(default)]
pub correlation_data: Option<String>,
/// 实际发送的字节。Base64/Hex 时为解码结果;不持久化、不参与预设。
/// None 时引擎回退为 `payload.as_bytes()`。
#[serde(skip)]
pub raw_bytes: Option<Vec<u8>>,
}
/// 发布预设:保存常用的发布参数,一键复用(对标 MQTTX Publish Scripts/Preset)。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PublishPreset {
pub id: String,
pub name: String,
#[serde(default)]
pub connection_id: Option<String>,
#[serde(flatten)]
pub params: PublishParams,
}
impl PublishPreset {
pub fn new(name: impl Into<String>, params: PublishParams) -> Self {
Self {
id: uuid::Uuid::new_v4().to_string(),
name: name.into(),
connection_id: None,
params,
}
}
}
// ─── 全局变量 ────────────────────────────────────────────────────────────────
+82
View File
@@ -0,0 +1,82 @@
//! 应用设置:连接状态枚举、主题偏好与全局设置项。
use serde::{Deserialize, Serialize};
use super::*;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ConnectionStatus {
#[default]
Disconnected,
Connecting,
Connected,
Error,
}
impl ConnectionStatus {
pub fn label(self) -> &'static str {
match self {
Self::Disconnected => "未连接",
Self::Connecting => "连接中",
Self::Connected => "已连接",
Self::Error => "错误",
}
}
}
// ─── 应用设置 ────────────────────────────────────────────────────────────────
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum ThemeModePref {
Light,
Dark,
#[default]
System,
}
impl ThemeModePref {
pub const ALL: [Self; 3] = [Self::System, Self::Light, Self::Dark];
pub fn label(self) -> &'static str {
match self {
Self::System => "跟随系统",
Self::Light => "浅色",
Self::Dark => "深色",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AppSettings {
#[serde(default)]
pub theme: ThemeModePref,
/// 每条连接在内存中保留的最大消息条数
#[serde(default = "default_max_messages")]
pub max_messages: usize,
/// 时间戳是否显示毫秒
#[serde(default = "default_true")]
pub show_millis: bool,
/// 是否自动检查更新(占位,暂不实现更新)
#[serde(default = "default_true")]
pub auto_check_update: bool,
}
fn default_max_messages() -> usize {
2000
}
impl Default for AppSettings {
fn default() -> Self {
Self {
theme: ThemeModePref::default(),
max_messages: default_max_messages(),
show_millis: true,
auto_check_update: true,
}
}
}
// ─── 单元测试 ────────────────────────────────────────────────────────────────
+80
View File
@@ -0,0 +1,80 @@
//! 订阅:订阅条目与下发引擎的订阅选项。
use serde::{Deserialize, Serialize};
use super::*;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Subscription {
pub id: String,
pub connection_id: String,
pub topic: String,
pub qos: u8,
/// 本地备注颜色(与 MQTTX 一致,按订阅给消息着色),0..=359 色相
#[serde(default)]
pub color: Option<f32>,
/// 订阅别名(对标 MQTTX topic alias),仅用于界面展示
#[serde(default)]
pub alias: Option<String>,
/// 是否启用该订阅;禁用后界面置灰、恢复订阅时跳过
#[serde(default = "default_true")]
pub enabled: bool,
/// MQTT 5 订阅标识符
#[serde(default)]
pub sub_identifier: Option<u32>,
/// MQTT 5 No Local (NL)
#[serde(default)]
pub no_local: bool,
/// MQTT 5 Retain As Published (RAP)
#[serde(default)]
pub retain_as_published: bool,
/// MQTT 5 Retain Handling (0/1/2)
#[serde(default)]
pub retain_handling: u8,
}
impl Subscription {
pub fn new(connection_id: impl Into<String>, topic: impl Into<String>, qos: u8) -> Self {
Self {
id: uuid::Uuid::new_v4().to_string(),
connection_id: connection_id.into(),
topic: topic.into(),
qos,
color: None,
alias: None,
enabled: true,
sub_identifier: None,
no_local: false,
retain_as_published: false,
retain_handling: 0,
}
}
}
/// MQTT 5 订阅选项(NL/RAP/Retain Handling/订阅标识符);v4 连接会忽略这些选项。
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SubscribeOptions {
/// 订阅标识符,0 视为未设置
#[serde(default)]
pub sub_identifier: Option<u32>,
#[serde(default)]
pub no_local: bool,
#[serde(default)]
pub retain_as_published: bool,
/// 0=每次订阅都发送保留消息,1=仅新订阅发送,2=不发送
#[serde(default)]
pub retain_handling: u8,
}
impl From<&Subscription> for SubscribeOptions {
fn from(sub: &Subscription) -> Self {
Self {
sub_identifier: sub.sub_identifier,
no_local: sub.no_local,
retain_as_published: sub.retain_as_published,
retain_handling: sub.retain_handling,
}
}
}
+71
View File
@@ -0,0 +1,71 @@
//! 全局变量:模板占位符渲染(含 {{timestamp}} 时间基一致性)。
use serde::{Deserialize, Serialize};
use super::*;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GlobalVariable {
pub key: String,
pub value: String,
}
/// 解析 `{{key}}` 模板:
/// - 用户定义的全局变量优先;
/// - 内置变量:`{{$ts}}`(unix 秒)、`{{$ts_ms}}`(毫秒)、`{{$uuid}}`;
/// - 找不到的占位符保持原样。
pub fn render_template(input: &str, vars: &[GlobalVariable]) -> String {
let mut out = String::with_capacity(input.len());
let bytes = input.as_bytes();
// 同一次渲染内时间基准保持一致,避免一条消息里多个时间变量取值不同。
let now_ms = chrono::Local::now().timestamp_millis();
let now_s = now_ms / 1000;
let mut i = 0;
while i < bytes.len() {
if i + 1 < bytes.len() && bytes[i] == b'{' && bytes[i + 1] == b'{'
&& let Some(rel_end) = input[i + 2..].find("}}") {
let key = &input[i + 2..i + 2 + rel_end];
let resolved = match key.trim() {
"$ts" => Some(now_s.to_string()),
"$ts_ms" => Some(now_ms.to_string()),
"$uuid" => Some(uuid::Uuid::new_v4().to_string()),
k => vars
.iter()
.find(|v| v.key == k)
.map(|v| v.value.clone()),
};
match resolved {
Some(v) => out.push_str(&v),
None => out.push_str(&input[i..i + 2 + rel_end + 2]),
}
i = i + 2 + rel_end + 2;
continue;
}
// 按字符推进,避免把多字节 UTF-8 拆开
let ch = input[i..].chars().next().unwrap();
out.push(ch);
i += ch.len_utf8();
}
out
}
/// 连接前渲染遗嘱消息中的 `{{变量}}`。
/// 只作用于本次连接的配置副本,落盘的连接配置保持模板原文,
/// 这样改全局变量后下次重连即生效,`$ts` 等也按连接时刻求值。
pub fn render_will_templates(cfg: &mut ConnectionConfig, vars: &[GlobalVariable]) {
let Some(will) = cfg.last_will.as_mut() else {
return;
};
will.topic = render_template(&will.topic, vars);
will.payload = render_template(&will.payload, vars);
if let Some(v) = will.content_type.take() {
will.content_type = Some(render_template(&v, vars));
}
if let Some(v) = will.response_topic.take() {
will.response_topic = Some(render_template(&v, vars));
}
}
// ─── 日志 ────────────────────────────────────────────────────────────────────
-1418
View File
File diff suppressed because it is too large Load Diff
+628
View File
@@ -0,0 +1,628 @@
//! MQTT 引擎:在独立 tokio runtime 上管理多个连接的生命周期,
//! 通过 smol 无界通道把状态/消息/日志事件送回 GPUI 线程。
//!
//! - [`tls`] TLS/rustls 配置与跳过校验
//! - [`v5`] MQTT 5.0 客户端构建与事件循环
//! - [`v4`] MQTT 3.1.1 客户端构建与事件循环
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use rumqttc::mqttbytes::v5::PublishProperties as V5PublishProperties;
use rumqttc::{
AsyncClient as V5AsyncClient, EventLoop as V5EventLoop,
PublishOptions as V5PublishOptions, RetainForwardRule, SubscribeFilterInput,
SubscribeProperties as V5SubscribeProperties,
};
use rumqttc_v4 as rv4;
use tokio::sync::watch;
use crate::model::{
ConnectionConfig, ConnectionStatus, LogEntry, LogLevel, MqttRecord, PublishParams, SubscribeOptions, Subscription,
};// ─── 引擎事件 ────────────────────────────────────────────────────────────────
mod tls;
mod v4;
mod v5;
use tls::*;
use v4::*;
use v5::*;
pub enum EngineEvent {
/// 连接状态变化
Status {
connection_id: String,
status: ConnectionStatus,
error: Option<String>,
reason_code: Option<u8>,
session_present: Option<bool>,
},
/// 收到消息
Message(MqttRecord),
/// 一条本地发出的消息(已成功交给客户端)。payload 为实际发送的字节。
/// 属性与接收记录对齐,本地发送记录可完整回显。
Published {
connection_id: String,
topic: String,
payload: Vec<u8>,
qos: u8,
retain: bool,
content_type: Option<String>,
user_properties: Vec<(String, String)>,
response_topic: Option<String>,
correlation_data: Option<String>,
message_expiry_interval: Option<u32>,
},
/// 订阅请求结果(tracked,等待 SUBACK)
SubscribeResult {
connection_id: String,
topic: String,
qos: u8,
ok: bool,
error: Option<String>,
},
/// 日志
Log(LogEntry),
}
pub type EventSender = smol::channel::Sender<EngineEvent>;
// ─── 句柄 ────────────────────────────────────────────────────────────────────
enum ClientKind {
V5(V5AsyncClient),
V4(rv4::AsyncClient),
}
struct ConnHandle {
/// `None` 表示仍在建立中(占位)。建连完成后由 connect 的任务升级为 Some。
client: Option<ClientKind>,
cancel_tx: watch::Sender<bool>,
connected: Arc<AtomicBool>,
/// 连接代数:同一 id 的第几次 connect。事件循环退出清理时只在
/// map 中仍是自己这一代时才移除,避免误删重连后的新句柄。
generation: u64,
}
pub struct MqttEngine {
runtime: tokio::runtime::Runtime,
conns: Mutex<HashMap<String, ConnHandle>>,
/// 下一次 connect 使用的代数
next_generation: AtomicU64,
/// 主题/连接 ID 驻留缓存:同一连接内主题高度重复,驻留让消息记录共享字符串
topics: Mutex<HashMap<String, Arc<str>>>,
tx: EventSender,
seq: Arc<AtomicU64>,
}
fn now_ms() -> i64 {
chrono::Local::now().timestamp_millis()
}
impl MqttEngine {
pub fn new(tx: EventSender) -> Arc<Self> {
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.thread_name("mqtt-engine")
.build()
.expect("创建 tokio runtime 失败");
Arc::new(Self {
runtime,
conns: Mutex::new(HashMap::new()),
next_generation: AtomicU64::new(1),
topics: Mutex::new(HashMap::new()),
tx,
seq: Arc::new(AtomicU64::new(1)),
})
}
fn next_seq(&self) -> u64 {
self.seq.fetch_add(1, Ordering::Relaxed)
}
fn emit(&self, event: EngineEvent) {
let _ = self.tx.try_send(event);
}
fn log(&self, conn_id: &str, level: LogLevel, event: &str, message: String, details: Option<String>) {
self.emit(EngineEvent::Log(LogEntry {
timestamp: now_ms(),
connection_id: conn_id.to_string(),
level,
event: event.to_string(),
message,
details,
}));
}
pub fn is_connected(&self, id: &str) -> bool {
let Ok(map) = self.conns.lock() else {
return false;
};
map.get(id)
.map(|h| h.connected.load(Ordering::SeqCst))
.unwrap_or(false)
}
// ── 连接 ──────────────────────────────────────────────────────────────
pub fn connect(self: &Arc<Self>, cfg: ConnectionConfig) {
// 关闭同名旧连接
self.close(&cfg.id, false);
let id = cfg.id.clone();
let generation = self.next_generation.fetch_add(1, Ordering::Relaxed) + 1;
let (cancel_tx, cancel_rx) = watch::channel(false);
let connected = Arc::new(AtomicBool::new(false));
// 同步占位(client=None):建连期间 close() 也能取消本次连接,
// 且 is_connected() 在建连完成前保持 false。
self.conns.lock().unwrap().insert(
id.clone(),
ConnHandle {
client: None,
cancel_tx,
connected: connected.clone(),
generation,
},
);
self.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Connecting,
error: None,
reason_code: None,
session_present: None,
});
self.log(
&id,
LogLevel::Info,
"connect",
format!(
"正在连接 {}:{} ({}, {})",
cfg.host,
cfg.port,
cfg.protocol.label(),
cfg.transport.label()
),
None,
);
let engine = Arc::clone(self);
self.runtime.spawn(async move {
let auto_reconnect = cfg.auto_reconnect;
let max_reconnect_times = cfg.max_reconnect_times;
// build 含证书/CA 文件读取等阻塞 IO,放 spawn_blocking,
// 避免慢盘上卡住 2 线程 runtime 里其他连接的调度
let build = tokio::task::spawn_blocking(move || {
if cfg.protocol.is_v5() {
build_v5(&cfg).map(|(client, el)| {
(ClientKind::V5(client), EventLoopKind::V5(Box::new(el)))
})
} else {
build_v4(&cfg).map(|(client, el)| {
(ClientKind::V4(client), EventLoopKind::V4(Box::new(el)))
})
}
})
.await;
let build_error = |engine: &Arc<MqttEngine>, id: &str, generation: u64, e: String| {
engine.remove_handle_if(id, generation);
engine.emit(EngineEvent::Status {
connection_id: id.to_string(),
status: ConnectionStatus::Error,
error: Some(e.clone()),
reason_code: None,
session_present: None,
});
engine.log(id, LogLevel::Error, "connect_error", format!("连接失败: {e}"), None);
};
let (client, eventloop) = match build {
Ok(Ok(pair)) => pair,
Ok(Err(e)) => {
build_error(&engine, &id, generation, e);
return;
}
Err(j) => {
build_error(&engine, &id, generation, format!("内部任务失败: {j}"));
return;
}
};
// 建连期间该连接可能已被 close 或更新的 connect 顶掉:
// 仅当占位句柄仍是本代时才升级为可用客户端,否则丢弃本次建连。
{
let mut map = match engine.conns.lock() {
Ok(m) => m,
Err(_) => return,
};
match map.get_mut(&id) {
Some(h) if h.generation == generation => h.client = Some(client),
_ => return,
}
}
match eventloop {
EventLoopKind::V5(el) => {
spawn_v5_loop(
engine.clone(),
id,
*el,
cancel_rx,
connected,
generation,
auto_reconnect,
max_reconnect_times,
);
}
EventLoopKind::V4(el) => {
spawn_v4_loop(
engine.clone(),
id,
*el,
cancel_rx,
connected,
generation,
auto_reconnect,
max_reconnect_times,
);
}
}
});
}
/// 主动断开。`notify` 为 true 时发出 Disconnected 状态(用户操作)。
pub fn close(&self, id: &str, user_initiated: bool) {
let handle = self.conns.lock().unwrap().remove(id);
if let Some(h) = handle {
let _ = h.cancel_tx.send(true);
let connected = h.connected.clone();
// 发送 MQTT DISCONNECT 报文后丢弃客户端(建连未完成时无需发送)
self.runtime.spawn(async move {
match h.client {
Some(ClientKind::V5(c)) => {
let _ = c.disconnect().await;
}
Some(ClientKind::V4(c)) => {
let _ = c.disconnect().await;
}
None => {}
}
connected.store(false, Ordering::SeqCst);
});
if user_initiated {
self.log(id, LogLevel::Info, "disconnect", "用户请求断开连接".into(), None);
self.emit(EngineEvent::Status {
connection_id: id.to_string(),
status: ConnectionStatus::Disconnected,
error: None,
reason_code: None,
session_present: None,
});
}
}
}
// ── 发布 ──────────────────────────────────────────────────────────────
pub fn publish(self: &Arc<Self>, connection_id: String, params: PublishParams) {
let engine = Arc::clone(self);
self.runtime.spawn(async move {
let Some(client) = engine.client_of(&connection_id) else {
engine.log(
&connection_id,
LogLevel::Error,
"publish_error",
"发布失败: 连接不存在".into(),
None,
);
return;
};
let payload = params
.raw_bytes
.clone()
.unwrap_or_else(|| params.payload.as_bytes().to_vec());
// 实际发送的字节留给 Published 事件(记录需要真实内容)
let sent_payload = payload.clone();
// v5 与 v4 两个 crate 各自定义 ClientError,统一归一化为 String。
let result: Result<(), String> = match &client {
ClientKind::V5(client) => {
let props = V5PublishProperties {
user_properties: params.user_properties.clone(),
content_type: params.content_type.clone(),
message_expiry_interval: params.message_expiry_interval,
response_topic: params.response_topic.clone(),
correlation_data: params
.correlation_data
.clone()
.map(|s| s.into_bytes().into()),
..Default::default()
};
let options = V5PublishOptions::new(v5_qos(params.qos))
.retain(params.retain)
.properties(props);
client
.publish(params.topic.clone(), payload, options)
.await
.map_err(|e| e.to_string())
}
ClientKind::V4(client) => {
let options =
rv4::PublishOptions::new(v4_qos(params.qos)).retain(params.retain);
client
.publish(params.topic.clone(), payload, options)
.await
.map_err(|e| e.to_string())
}
};
match result {
Ok(_) => {
engine.log(
&connection_id,
LogLevel::Info,
"publish_sent",
format!(
"发布消息到 {} (QoS {}, retain={})",
params.topic, params.qos, params.retain
),
None,
);
engine.emit(EngineEvent::Published {
connection_id,
topic: params.topic,
payload: sent_payload,
qos: params.qos,
retain: params.retain,
content_type: params.content_type,
user_properties: params.user_properties,
response_topic: params.response_topic,
correlation_data: params.correlation_data,
message_expiry_interval: params.message_expiry_interval,
});
}
Err(e) => {
engine.log(
&connection_id,
LogLevel::Error,
"publish_error",
format!("发布失败: {e}"),
None,
);
}
}
});
}
// ── 订阅 / 取消订阅 ─────────────────────────────────────────────────────
/// 旧签名薄封装:不携带 v5 订阅选项(连接视图等现有调用点继续可用)。
pub fn subscribe(self: &Arc<Self>, connection_id: String, sub: Subscription) {
self.subscribe_with_options(connection_id, sub, SubscribeOptions::default());
}
/// 带 MQTT 5 订阅选项的订阅(sub_identifier / NL / RAP / Retain Handling);
/// v4 连接会忽略这些选项。
pub fn subscribe_with_options(
self: &Arc<Self>,
connection_id: String,
sub: Subscription,
opts: SubscribeOptions,
) {
let engine = Arc::clone(self);
self.runtime.spawn(async move {
let Some(client) = engine.client_of(&connection_id) else {
engine.emit(EngineEvent::SubscribeResult {
connection_id,
topic: sub.topic,
qos: sub.qos,
ok: false,
error: Some("连接不存在".into()),
});
return;
};
engine.log(
&connection_id,
LogLevel::Info,
"subscribe",
format!("订阅主题: {} (QoS {})", sub.topic, sub.qos),
None,
);
let result = match &client {
ClientKind::V5(client) => {
// NL/RAP/Retain Handling 是「按主题过滤器」的订阅选项,
// 订阅标识符则在订阅报文属性里,分别写入两处。
let filter = SubscribeFilterInput::new(sub.topic.clone(), v5_qos(sub.qos))
.no_local(opts.no_local)
.preserve_retain(opts.retain_as_published)
.retain_forward_rule(match opts.retain_handling {
1 => RetainForwardRule::OnNewSubscribe,
2 => RetainForwardRule::Never,
_ => RetainForwardRule::OnEverySubscribe,
});
let properties = V5SubscribeProperties {
// 0 不是合法的订阅标识符,按未设置处理
id: opts.sub_identifier.filter(|v| *v > 0).map(|v| v as usize),
user_properties: Vec::new(),
};
match client
.subscribe_many_with_properties_tracked(std::iter::once(filter), properties)
.await
{
Ok(notice) => notice.wait_completion_async().await.map_err(|e| {
format!("订阅被拒绝: {e}")
}),
Err(e) => Err(format!("订阅请求失败: {e}")),
}
}
ClientKind::V4(client) => {
match client
.subscribe_tracked(sub.topic.clone(), v4_qos(sub.qos))
.await
{
Ok(notice) => notice.wait_completion_async().await.map_err(|e| {
format!("订阅被拒绝: {e}")
}),
Err(e) => Err(format!("订阅请求失败: {e}")),
}
}
};
let (ok, error) = match result {
Ok(()) => (true, None),
Err(e) => {
engine.log(
&connection_id,
LogLevel::Error,
"subscribe_error",
e.clone(),
None,
);
(false, Some(e))
}
};
engine.emit(EngineEvent::SubscribeResult {
connection_id,
topic: sub.topic,
qos: sub.qos,
ok,
error,
});
});
}
pub fn unsubscribe(self: &Arc<Self>, connection_id: String, topic: String) {
let engine = Arc::clone(self);
self.runtime.spawn(async move {
let Some(client) = engine.client_of(&connection_id) else {
return;
};
let result: Result<(), String> = match &client {
ClientKind::V5(client) => client
.unsubscribe(topic.clone())
.await
.map_err(|e| e.to_string()),
ClientKind::V4(client) => client
.unsubscribe(topic.clone())
.await
.map_err(|e| e.to_string()),
};
match result {
Ok(()) => {
engine.log(
&connection_id,
LogLevel::Info,
"unsubscribe",
format!("取消订阅: {topic}"),
None,
);
}
Err(e) => {
engine.log(
&connection_id,
LogLevel::Error,
"unsubscribe_error",
format!("取消订阅失败: {e}"),
None,
);
}
}
});
}
/// 测试连接:使用配置的连接超时(上限 30 秒,防止误填过大值长时间挂起),
/// 结果通过 channel 返回。
/// 在引擎的 tokio runtime 上调度阻塞任务(OTA 下载等),
/// 结果经 smol 通道送回 GPUI 执行器;任务 panic 时接收端以空错误结束。
pub fn run_blocking<T: Send + 'static>(
&self,
f: impl FnOnce() -> T + Send + 'static,
) -> smol::channel::Receiver<T> {
let (tx, rx) = smol::channel::bounded(1);
self.runtime.spawn(async move {
if let Ok(value) = tokio::task::spawn_blocking(f).await {
let _ = tx.send(value).await;
}
});
rx
}
/// 驻留字符串(主题/连接 ID):同一内容全进程只保留一份 `Arc<str>`。
/// 缓存有上限,异常流量下整体清空重建,避免无限增长。
pub(crate) fn intern(&self, s: &str) -> Arc<str> {
const INTERN_CAP: usize = 10_000;
let mut cache = self.topics.lock().unwrap();
if let Some(arc) = cache.get(s) {
return arc.clone();
}
if cache.len() >= INTERN_CAP {
cache.clear();
}
let arc: Arc<str> = Arc::from(s);
cache.insert(s.to_string(), arc.clone());
arc
}
pub fn test_connection(
self: &Arc<Self>,
cfg: ConnectionConfig,
) -> smol::channel::Receiver<Result<(), String>> {
let (tx, rx) = smol::channel::bounded(1);
let secs = u64::from(cfg.connection_timeout_secs.clamp(1, 30));
self.runtime.spawn(async move {
let result = tokio::time::timeout(Duration::from_secs(secs), async {
test_handshake(&cfg).await
})
.await;
let reply = match result {
Ok(inner) => inner,
Err(_) => Err(format!(
"连接测试超时({secs} 秒),请检查主机地址与端口"
)),
};
let _ = tx.try_send(reply);
});
rx
}
// ── 内部工具 ───────────────────────────────────────────────────────────
/// 克隆一个连接的异步客户端用于发命令。
/// 不持有 std Mutex 守卫跨 await,避免与 tokio runtime 的 Send 要求冲突。
/// 占位句柄(建连未完成)视为不存在。
fn client_of(&self, id: &str) -> Option<ClientKind> {
let map = self.conns.lock().ok()?;
let client = map.get(id)?.client.as_ref()?;
Some(match client {
ClientKind::V5(c) => ClientKind::V5(c.clone()),
ClientKind::V4(c) => ClientKind::V4(c.clone()),
})
}
/// 仅当 map 中该 id 的句柄仍是 `generation` 这一代时才移除。
/// 事件循环退出清理必须走这里:无条件按 id 删除会把重连后的
/// 新句柄误删,导致连接无法再从 UI 断开。
fn remove_handle_if(&self, id: &str, generation: u64) {
if let Ok(mut map) = self.conns.lock()
&& map.get(id).map(|h| h.generation) == Some(generation)
{
map.remove(id);
}
}
}
enum EventLoopKind {
// 两个事件循环都在 KB 级,装箱让枚举本身只有指针大小
V5(Box<V5EventLoop>),
V4(Box<rv4::EventLoop>),
}
// ─── 选项构建 ────────────────────────────────────────────────────────────────
/// 连接超时秒数:至少 1 秒,避免误填 0 导致立刻超时。
fn connection_timeout_secs(cfg: &ConnectionConfig) -> u64 {
u64::from(cfg.connection_timeout_secs.max(1))
}
+178
View File
@@ -0,0 +1,178 @@
//! TLS:由 SslConfig 解析 rustls 配置,含跳过证书校验的 NoVerifier。
use std::sync::Arc;
use rumqttc::TlsConfiguration;
use rustls::pki_types::{pem::PemObject, CertificateDer, PrivateKeyDer};
use crate::model::{
ConnectionConfig,
SslConfig,
};// ─── 引擎事件 ────────────────────────────────────────────────────────────────
/// SSL 配置非默认时构建自定义 rustls 配置;返回 None 表示沿用 rumqttc 默认 TLS。
/// 文件读取/解析失败在此返回 Err,由 connect 的错误通道上报为连接失败事件。
pub(super) fn resolve_tls_config(cfg: &ConnectionConfig) -> Result<Option<TlsConfiguration>, String> {
if !cfg.transport.is_tls() || cfg.ssl.is_default() {
return Ok(None);
}
Ok(Some(TlsConfiguration::Rustls(build_rustls_config(&cfg.ssl)?)))
}
/// 按 SslConfig 构造 rustls ClientConfig:
/// - `ignore_ca`:跳过服务器证书链/主机名校验;
/// - `ca_file`:在系统根证书之外追加自定义 CA;
/// - `client_cert_file` + `client_key_file`:双向认证。
pub(super) fn build_rustls_config(ssl: &SslConfig) -> Result<Arc<rustls::ClientConfig>, String> {
// 只填证书或只填私钥必然握手失败,提前给出明确错误
if ssl.client_cert_file.is_empty() != ssl.client_key_file.is_empty() {
return Err("客户端证书与客户端密钥必须同时配置".into());
}
let provider = ensure_crypto_provider();
let builder = rustls::ClientConfig::builder();
let builder = if ssl.ignore_ca {
builder
.dangerous()
.with_custom_certificate_verifier(Arc::new(NoVerifier {
provider: provider.clone(),
}))
} else {
let mut roots = rustls::RootCertStore::empty();
let native = rustls_native_certs::load_native_certs();
if native.certs.is_empty() && ssl.ca_file.is_empty() {
let detail = native
.errors
.iter()
.map(|e| e.to_string())
.collect::<Vec<_>>()
.join("; ");
return Err(format!("加载系统根证书失败: {detail}"));
}
// 个别系统证书加载失败不阻断连接:自定义 CA 与其余根证书仍参与校验
for cert in native.certs {
let _ = roots.add(cert);
}
if !ssl.ca_file.is_empty() {
let pem = std::fs::read(&ssl.ca_file)
.map_err(|e| format!("读取 CA 文件 {} 失败: {e}", ssl.ca_file))?;
let mut added = 0usize;
for cert in CertificateDer::pem_slice_iter(&pem) {
let cert = cert.map_err(|e| format!("解析 CA 文件 {} 失败: {e}", ssl.ca_file))?;
roots
.add(cert)
.map_err(|e| format!("CA 文件 {} 中存在无效证书: {e}", ssl.ca_file))?;
added += 1;
}
if added == 0 {
return Err(format!("CA 文件 {} 中没有找到任何证书", ssl.ca_file));
}
}
builder.with_root_certificates(roots)
};
let config = if ssl.client_cert_file.is_empty() {
builder.with_no_client_auth()
} else {
let chain_pem = std::fs::read(&ssl.client_cert_file).map_err(|e| {
format!("读取客户端证书文件 {} 失败: {e}", ssl.client_cert_file)
})?;
let key_pem = std::fs::read(&ssl.client_key_file)
.map_err(|e| format!("读取客户端私钥文件 {} 失败: {e}", ssl.client_key_file))?;
let certs = CertificateDer::pem_slice_iter(&chain_pem)
.collect::<Result<Vec<_>, _>>()
.map_err(|e| format!("解析客户端证书文件 {} 失败: {e}", ssl.client_cert_file))?;
if certs.is_empty() {
return Err(format!(
"客户端证书文件 {} 中没有找到证书",
ssl.client_cert_file
));
}
let key = PrivateKeyDer::from_pem_slice(&key_pem)
.map_err(|e| format!("解析客户端私钥文件 {} 失败: {e}", ssl.client_key_file))?;
builder
.with_client_auth_cert(certs, key)
.map_err(|e| format!("加载客户端证书/私钥失败: {e}"))?
};
Ok(Arc::new(config))
}
/// 取得(必要时安装)进程级 CryptoProvider。
/// 工作区同时启用 ring 与 aws-lc-rs 特性时 rustls 无法自动推断默认 provider,
/// 不显式安装会让 `ClientConfig::builder()` panic。
pub(super) fn ensure_crypto_provider() -> Arc<rustls::crypto::CryptoProvider> {
if let Some(provider) = rustls::crypto::CryptoProvider::get_default() {
return Arc::clone(provider);
}
let provider = rustls::crypto::aws_lc_rs::default_provider();
// 输掉安装竞争时直接采用已安装的那个
match provider.clone().install_default() {
Ok(()) => Arc::new(provider),
Err(installed) => installed,
}
}
/// 「忽略 CA 校验」用:跳过服务器证书链与主机名校验,
/// 握手签名仍用 provider 校验(与 rustls 官方示例做法一致)。
pub(super) struct NoVerifier {
provider: Arc<rustls::crypto::CryptoProvider>,
}
impl std::fmt::Debug for NoVerifier {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("NoVerifier").finish_non_exhaustive()
}
}
impl rustls::client::danger::ServerCertVerifier for NoVerifier {
fn verify_server_cert(
&self,
_end_entity: &CertificateDer<'_>,
_intermediates: &[CertificateDer<'_>],
_server_name: &rustls::pki_types::ServerName<'_>,
_ocsp_response: &[u8],
_now: rustls::pki_types::UnixTime,
) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
Ok(rustls::client::danger::ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
message: &[u8],
cert: &CertificateDer<'_>,
dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
rustls::crypto::verify_tls12_signature(
message,
cert,
dss,
&self.provider.signature_verification_algorithms,
)
}
fn verify_tls13_signature(
&self,
message: &[u8],
cert: &CertificateDer<'_>,
dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
rustls::crypto::verify_tls13_signature(
message,
cert,
dss,
&self.provider.signature_verification_algorithms,
)
}
fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
self.provider
.signature_verification_algorithms
.supported_schemes()
}
}
/// 去掉空白字符串,避免表单存了空串也写进报文属性。
pub(super) fn non_blank(s: Option<String>) -> Option<String> {
s.filter(|v| !v.trim().is_empty())
}
+317
View File
@@ -0,0 +1,317 @@
//! MQTT 3.1.1:MqttOptions/last-will 组装与事件循环泵。
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use rumqttc::mqttbytes::v5::{
ConnectReturnCode as V5ReturnCode, Packet as V5Packet,
};
use rumqttc::Event as V5Event;
use rumqttc_v4 as rv4;
use tokio::sync::watch;
use crate::model::{
ConnectionConfig, ConnectionStatus, Direction, LogLevel, MqttRecord, TransportKind,
};// ─── 引擎事件 ────────────────────────────────────────────────────────────────
use super::*;
pub(super) fn build_v4(
cfg: &ConnectionConfig,
) -> Result<(rv4::AsyncClient, rv4::EventLoop), String> {
use rv4::{Broker as V4Broker, MqttOptions as V4Options, Transport as V4Transport};
let custom_tls = resolve_tls_config(cfg)?;
let mut opts = match cfg.transport {
TransportKind::Tcp | TransportKind::Tls => {
V4Options::new(cfg.client_id.clone(), V4Broker::tcp(&cfg.host, cfg.port))
}
TransportKind::Ws => {
let url = format!("ws://{}:{}{}", cfg.host, cfg.port, normalize_path(&cfg.path));
V4Options::new(cfg.client_id.clone(), V4Broker::websocket(url).map_err(|e| format!("{e:?}"))?)
}
TransportKind::Wss => {
let url = format!("wss://{}:{}{}", cfg.host, cfg.port, normalize_path(&cfg.path));
match &custom_tls {
Some(tls) => V4Options::websocket_with_tls_config(
cfg.client_id.clone(),
url,
tls.clone(),
)
.map_err(|e| format!("{e:?}"))?,
None => V4Options::try_websocket_with_default_tls(cfg.client_id.clone(), url)
.map_err(|e| format!("{e:?}"))?,
}
}
};
if cfg.transport == TransportKind::Tls {
match &custom_tls {
Some(tls) => {
opts.set_transport(V4Transport::tls_with_config(tls.clone()));
}
None => {
opts.set_transport(
V4Transport::try_tls_with_default_config()
.map_err(|e| format!("TLS 初始化失败: {e:?}"))?,
);
}
};
}
opts.set_keep_alive(cfg.keep_alive);
opts.set_clean_session(cfg.clean_start);
if !cfg.username.is_empty() {
opts.set_credentials(cfg.username.clone(), cfg.password.clone());
}
if let Some(will) = &cfg.last_will
&& !will.topic.is_empty() {
// 3.1.1 的遗嘱没有 content_type / response_topic 属性,忽略
let lw = rv4::mqttbytes::v4::LastWill::new(
will.topic.clone(),
will.payload.as_bytes().to_vec(),
v4_qos(will.qos),
will.retain,
);
opts.set_last_will(lw);
}
let (client, mut eventloop) = rv4::AsyncClient::builder(opts).build();
// v4 的连接超时(含 CONNACK 等待)由 EventLoop 的 NetworkOptions 控制
let mut network_options = rv4::NetworkOptions::new();
network_options.set_connection_timeout(connection_timeout_secs(cfg));
eventloop.set_network_options(network_options);
Ok((client, eventloop))
}
pub(super) fn v4_qos(qos: u8) -> rv4::QoS {
match qos {
1 => rv4::QoS::AtLeastOnce,
2 => rv4::QoS::ExactlyOnce,
_ => rv4::QoS::AtMostOnce,
}
}
// ─── 测试连接握手 ────────────────────────────────────────────────────────────
pub(super) async fn test_handshake(cfg_in: &ConnectionConfig) -> Result<(), String> {
// 握手测试必须使用独立 client_id,否则会把使用相同 client_id 的现有连接
// 从 broker 顶下线(MQTT 同 client_id 单会话规则)。
let mut cfg = cfg_in.clone();
cfg.client_id = format!("mqttx_test_{}", &uuid::Uuid::new_v4().simple().to_string()[..8]);
let cfg = &cfg;
if cfg.protocol.is_v5() {
let (client, mut eventloop) = build_v5(cfg)?;
loop {
match eventloop.poll().await {
Ok(V5Event::Incoming(V5Packet::ConnAck(ack))) => {
if ack.code == V5ReturnCode::Success {
// 测试完成,主动断开避免在 broker 侧残留会话。
// disconnect() 只是把请求入队,需再 poll 才会把
// DISCONNECT 真正写上网络。
let _ = client.disconnect().await;
for _ in 0..3 {
if eventloop.poll().await.is_err() {
break;
}
}
return Ok(());
}
return Err(format!("连接被拒绝: {:?}", ack.code));
}
Ok(V5Event::Incoming(V5Packet::Disconnect(d))) => {
return Err(format!("服务端断开: {:?}", d.reason_code));
}
Err(e) => return Err(format!("连接失败: {e}")),
_ => {}
}
}
} else {
let (client, mut eventloop) = build_v4(cfg)?;
loop {
match eventloop.poll().await {
Ok(rv4::Event::Incoming(rv4::Packet::ConnAck(ack))) => {
if ack.code == rv4::ConnectReturnCode::Success {
// 同 v5:disconnect() 只入队,需 poll 把报文冲出
let _ = client.disconnect().await;
for _ in 0..3 {
if eventloop.poll().await.is_err() {
break;
}
}
return Ok(());
}
return Err(format!("连接被拒绝: {:?}", ack.code));
}
Ok(rv4::Event::Incoming(rv4::Packet::Disconnect)) => {
return Err("服务端主动断开".into());
}
Err(e) => return Err(format!("连接失败: {e}")),
_ => {}
}
}
}
}
// ─── 事件循环 ────────────────────────────────────────────────────────────────
pub(super) fn spawn_v4_loop(
engine: Arc<MqttEngine>,
id: String,
mut eventloop: rv4::EventLoop,
mut cancel_rx: watch::Receiver<bool>,
connected: Arc<AtomicBool>,
generation: u64,
auto_reconnect: bool,
max_reconnect_times: u32,
) {
#![allow(clippy::too_many_arguments)]
let runtime = engine.runtime.handle().clone();
runtime.spawn(async move {
let mut first_connect = true;
// 已执行的重连次数,用于 max_reconnect_times(0=无限)限流
let mut reconnect_attempts: u32 = 0;
loop {
tokio::select! {
event = eventloop.poll() => {
match event {
Ok(rv4::Event::Incoming(rv4::Packet::Publish(p))) => {
let (payload, raw_bytes, payload_truncated) =
crate::model::retained_payload(&p.payload);
let record = MqttRecord {
seq: engine.next_seq(),
connection_id: engine.intern(&id),
direction: Direction::Received,
topic: engine.intern(&String::from_utf8_lossy(&p.topic)),
payload,
qos: match p.qos {
rv4::QoS::AtMostOnce => 0,
rv4::QoS::AtLeastOnce => 1,
rv4::QoS::ExactlyOnce => 2,
},
retain: p.retain,
timestamp: now_ms(),
user_properties: Vec::new(),
content_type: None,
response_topic: None,
correlation_data: None,
message_expiry_interval: None,
subscription_identifier: None,
payload_truncated,
raw_bytes,
};
engine.log(
&id, LogLevel::Info, "publish_received",
format!("收到消息: {} ({} 字节)", record.topic, p.payload.len()), None);
engine.emit(EngineEvent::Message(record));
}
Ok(rv4::Event::Incoming(rv4::Packet::ConnAck(ack))) => {
connected.store(ack.code == rv4::ConnectReturnCode::Success, Ordering::SeqCst);
if ack.code == rv4::ConnectReturnCode::Success {
first_connect = false;
// 重连成功后清零计数:max_reconnect_times 限流的是
// 「连续失败次数」,不能把历史错误累计进来
reconnect_attempts = 0;
engine.log(&id, LogLevel::Info, "connack",
format!("连接成功 (session_present: {})", ack.session_present), None);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Connected,
error: None,
reason_code: Some(0),
session_present: Some(ack.session_present),
});
} else {
engine.log(&id, LogLevel::Error, "connack",
format!("连接被拒绝: {:?}", ack.code), None);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Error,
error: Some(format!("{:?}", ack.code)),
// 上报真实的 CONNACK 拒绝码,不再硬编码为 1
reason_code: Some(ack.code as u8),
session_present: Some(ack.session_present),
});
break;
}
}
Ok(rv4::Event::Incoming(rv4::Packet::Disconnect)) => {
connected.store(false, Ordering::SeqCst);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Disconnected,
error: Some("服务端断开".into()),
reason_code: None,
session_present: None,
});
break;
}
Ok(rv4::Event::Incoming(rv4::Packet::SubAck(s))) => {
engine.log(&id, LogLevel::Info, "suback",
format!("订阅确认 (packet {})", s.pkid), None);
}
Ok(rv4::Event::Incoming(rv4::Packet::UnsubAck(a))) => {
engine.log(&id, LogLevel::Info, "unsuback",
format!("取消订阅确认 (packet {})", a.pkid), None);
}
Ok(rv4::Event::Incoming(rv4::Packet::PubAck(a))) => {
engine.log(&id, LogLevel::Debug, "puback",
format!("发布确认 (packet {})", a.pkid), None);
}
Ok(rv4::Event::Outgoing(rv4::Outgoing::Publish(pkid))) => {
engine.log(&id, LogLevel::Debug, "publish_outgoing",
format!("消息已发出 (packet {pkid})"), None);
}
Ok(_) => {}
Err(rv4::ConnectionError::RequestsDone) => {
connected.store(false, Ordering::SeqCst);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Disconnected,
error: None, reason_code: None, session_present: None,
});
break;
}
Err(e) => {
connected.store(false, Ordering::SeqCst);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Error,
error: Some(e.to_string()),
reason_code: None, session_present: None,
});
if !auto_reconnect || first_connect {
engine.log(&id, LogLevel::Error, "connection_error",
format!("连接错误: {e}"), None);
break;
}
// max_reconnect_times>0 时按次数停止重试;0 表示无限重连
if max_reconnect_times > 0 && reconnect_attempts >= max_reconnect_times {
engine.log(&id, LogLevel::Error, "reconnect_give_up",
format!("已重连 {reconnect_attempts} 次仍失败,达到最大重连次数 {max_reconnect_times},停止重连: {e}"), None);
break;
}
reconnect_attempts += 1;
engine.log(&id, LogLevel::Warn, "connection_error",
format!("连接错误,2 秒后重连(第 {reconnect_attempts} 次): {e}"), None);
// 退避等待也必须可取消:否则 close() 的取消信号最长
// 2 秒不可见,期间会对用户已关闭的连接继续重连。
tokio::select! {
_ = cancel_rx.changed() => break,
_ = tokio::time::sleep(Duration::from_secs(2)) => {}
}
}
}
}
_ = cancel_rx.changed() => {
engine.log(&id, LogLevel::Info, "disconnect_cancelled", "连接已取消".into(), None);
break;
}
}
}
// 仅当 map 中仍是本代句柄时才移除,避免误删重连后的新句柄
engine.remove_handle_if(&id, generation);
});
}
+354
View File
@@ -0,0 +1,354 @@
//! MQTT 5.0:MqttOptions/last-will/属性组装与事件循环泵。
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use rumqttc::mqttbytes::v5::{
ConnectReturnCode as V5ReturnCode, Packet as V5Packet, Publish as V5Publish,
};
use rumqttc::{
AsyncClient as V5AsyncClient, Broker as V5Broker, ConnectionError as V5Error,
Event as V5Event, EventLoop as V5EventLoop, MqttOptions as V5Options, Outgoing as V5Outgoing, QoS as V5QoS, Transport as V5Transport,
};
use tokio::sync::watch;
use crate::model::{
ConnectionConfig, ConnectionStatus, Direction, LogLevel, MqttRecord, TransportKind,
};// ─── 引擎事件 ────────────────────────────────────────────────────────────────
use super::*;
pub(super) fn build_v5(
cfg: &ConnectionConfig,
) -> Result<(V5AsyncClient, V5EventLoop), String> {
let custom_tls = resolve_tls_config(cfg)?;
let mut opts = match cfg.transport {
TransportKind::Tcp | TransportKind::Tls => {
V5Options::new(cfg.client_id.clone(), V5Broker::tcp(&cfg.host, cfg.port))
}
TransportKind::Ws => {
let url = format!("ws://{}:{}{}", cfg.host, cfg.port, normalize_path(&cfg.path));
V5Options::new(cfg.client_id.clone(), V5Broker::websocket(url).map_err(|e| format!("{e:?}"))?)
}
TransportKind::Wss => {
let url = format!("wss://{}:{}{}", cfg.host, cfg.port, normalize_path(&cfg.path));
match &custom_tls {
Some(tls) => V5Options::websocket_with_tls_config(
cfg.client_id.clone(),
url,
tls.clone(),
)
.map_err(|e| format!("{e:?}"))?,
None => V5Options::try_websocket_with_default_tls(cfg.client_id.clone(), url)
.map_err(|e| format!("{e:?}"))?,
}
}
};
if cfg.transport == TransportKind::Tls {
match &custom_tls {
Some(tls) => {
opts.set_transport(V5Transport::tls_with_config(tls.clone()));
}
None => {
opts.set_transport(
V5Transport::try_tls_with_default_config()
.map_err(|e| format!("TLS 初始化失败: {e:?}"))?,
);
}
};
}
// 连接超时同时覆盖底层建连与 CONNACK 等待
let timeout = Duration::from_secs(connection_timeout_secs(cfg));
opts.set_connect_timeout(timeout);
let mut network_options = rumqttc::NetworkOptions::new();
network_options.set_connection_timeout(connection_timeout_secs(cfg));
opts.set_network_options(network_options);
opts.set_keep_alive(cfg.keep_alive);
opts.set_clean_start(cfg.clean_start);
if cfg.session_expiry_interval > 0 {
opts.set_session_expiry_interval(Some(cfg.session_expiry_interval));
}
if !cfg.username.is_empty() {
opts.set_credentials(cfg.username.clone(), cfg.password.clone());
}
if let Some(v) = cfg.receive_maximum {
opts.set_receive_maximum(Some(v));
}
if let Some(v) = cfg.maximum_packet_size {
opts.set_max_packet_size(Some(v));
}
if let Some(v) = cfg.topic_alias_maximum {
opts.set_topic_alias_max(Some(v));
}
if let Some(will) = &cfg.last_will
&& !will.topic.is_empty() {
// v5 遗嘱属性:content_type / response_topic(3.1.1 下无此扩展)
let properties = if will.content_type.is_some() || will.response_topic.is_some() {
Some(rumqttc::mqttbytes::v5::LastWillProperties {
delay_interval: None,
payload_format_indicator: None,
message_expiry_interval: None,
content_type: non_blank(will.content_type.clone()),
response_topic: non_blank(will.response_topic.clone()),
correlation_data: None,
user_properties: Vec::new(),
})
} else {
None
};
let lw = rumqttc::mqttbytes::v5::LastWill::new(
will.topic.clone(),
will.payload.as_bytes().to_vec(),
v5_qos(will.qos),
will.retain,
properties,
);
opts.set_last_will(lw);
}
Ok(V5AsyncClient::builder(opts).build())
}
pub(super) fn normalize_path(path: &str) -> String {
if path.is_empty() {
"/mqtt".to_string()
} else if path.starts_with('/') {
path.to_string()
} else {
format!("/{path}")
}
}
pub(super) fn v5_qos(qos: u8) -> V5QoS {
match qos {
1 => V5QoS::AtLeastOnce,
2 => V5QoS::ExactlyOnce,
_ => V5QoS::AtMostOnce,
}
}
fn v5_record(engine: &MqttEngine, seq: u64, conn_id: &str, p: &V5Publish) -> MqttRecord {
let props = p.properties.as_ref();
let (payload, raw_bytes, payload_truncated) = crate::model::retained_payload(&p.payload);
MqttRecord {
seq,
connection_id: engine.intern(conn_id),
direction: Direction::Received,
topic: engine.intern(&String::from_utf8_lossy(&p.topic)),
payload,
qos: v5_qos_u8(p.qos),
retain: p.retain,
timestamp: now_ms(),
user_properties: props
.map(|p| p.user_properties.clone())
.unwrap_or_default(),
content_type: props.and_then(|p| p.content_type.clone()),
response_topic: props.and_then(|p| p.response_topic.clone()),
correlation_data: props.and_then(|p| {
p.correlation_data
.as_ref()
.map(|d| correlation_display(d))
}),
message_expiry_interval: props.and_then(|p| p.message_expiry_interval),
subscription_identifier: props
.and_then(|p| p.subscription_identifiers.first().map(|v| *v as u32)),
payload_truncated,
raw_bytes,
}
}
/// correlation data 的可读表示:优先按 UTF-8 文本展示,非文本回退 hex。
/// 发送端把字符串按字节原样发出,两侧展示保持对称(此前收端一律 hex、
/// 发端保留原文,同一条数据两侧对不上)。
fn correlation_display(data: &[u8]) -> String {
match std::str::from_utf8(data) {
Ok(s) => s.to_string(),
Err(_) => data.iter().map(|b| format!("{b:02x}")).collect(),
}
}
fn v5_qos_u8(qos: V5QoS) -> u8 {
match qos {
V5QoS::AtMostOnce => 0,
V5QoS::AtLeastOnce => 1,
V5QoS::ExactlyOnce => 2,
}
}
pub(super) fn spawn_v5_loop(
engine: Arc<MqttEngine>,
id: String,
mut eventloop: V5EventLoop,
mut cancel_rx: watch::Receiver<bool>,
connected: Arc<AtomicBool>,
generation: u64,
auto_reconnect: bool,
max_reconnect_times: u32,
) {
#![allow(clippy::too_many_arguments)]
let runtime = engine.runtime.handle().clone();
runtime.spawn(async move {
let mut first_connect = true;
// 已执行的重连次数,用于 max_reconnect_times(0=无限)限流
let mut reconnect_attempts: u32 = 0;
loop {
tokio::select! {
event = eventloop.poll() => {
match event {
Ok(V5Event::Incoming(V5Packet::Publish(p))) => {
let topic = String::from_utf8_lossy(&p.topic).to_string();
let record = v5_record(&engine, engine.next_seq(), &id, &p);
engine.log(
&id,
LogLevel::Info,
"publish_received",
format!("收到消息: {topic} ({} 字节, QoS {})", p.payload.len(), v5_qos_u8(p.qos)),
None,
);
engine.emit(EngineEvent::Message(record));
}
Ok(V5Event::Incoming(V5Packet::ConnAck(ack))) => {
let code = ack.code as u8;
connected.store(ack.code == V5ReturnCode::Success, Ordering::SeqCst);
if ack.code == V5ReturnCode::Success {
first_connect = false;
// 重连成功后清零计数:max_reconnect_times 限流的是
// 「连续失败次数」,不能把历史错误累计进来
reconnect_attempts = 0;
engine.log(
&id,
LogLevel::Info,
"connack",
format!("连接成功 (session_present: {})", ack.session_present),
None,
);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Connected,
error: None,
reason_code: Some(code),
session_present: Some(ack.session_present),
});
} else {
engine.log(
&id,
LogLevel::Error,
"connack",
format!("连接被拒绝: {:?} (code {code})", ack.code),
None,
);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Error,
error: Some(format!("{:?}", ack.code)),
reason_code: Some(code),
session_present: Some(ack.session_present),
});
break;
}
}
Ok(V5Event::Incoming(V5Packet::Disconnect(d))) => {
connected.store(false, Ordering::SeqCst);
engine.log(
&id,
LogLevel::Warn,
"disconnect_received",
format!("收到断开连接: {:?} (code {})", d.reason_code, d.reason_code as u8),
None,
);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Disconnected,
error: Some(format!("{:?}", d.reason_code)),
reason_code: Some(d.reason_code as u8),
session_present: None,
});
break;
}
Ok(V5Event::Incoming(V5Packet::SubAck(s))) => {
let granted: Vec<String> = s
.return_codes
.iter()
.map(|c| format!("{c:?}"))
.collect();
engine.log(&id, LogLevel::Info, "suback",
format!("订阅确认 (packet {}): {}", s.pkid, granted.join(", ")), None);
}
Ok(V5Event::Incoming(V5Packet::UnsubAck(_))) => {
engine.log(&id, LogLevel::Info, "unsuback", "取消订阅确认".into(), None);
}
Ok(V5Event::Incoming(V5Packet::PubAck(a))) => {
engine.log(&id, LogLevel::Debug, "puback",
format!("发布确认 (packet {}, reason {:?})", a.pkid, a.reason), None);
}
Ok(V5Event::Incoming(V5Packet::PubRec(a))) => {
engine.log(&id, LogLevel::Debug, "pubrec",
format!("QoS2 发布收到 (packet {})", a.pkid), None);
}
Ok(V5Event::Incoming(V5Packet::PubComp(a))) => {
engine.log(&id, LogLevel::Debug, "pubcomp",
format!("QoS2 发布完成 (packet {})", a.pkid), None);
}
Ok(V5Event::Outgoing(V5Outgoing::Publish(pkid))) => {
engine.log(&id, LogLevel::Debug, "publish_outgoing",
format!("消息已发出 (packet {pkid})"), None);
}
Ok(_) => {}
Err(V5Error::RequestsDone) => {
connected.store(false, Ordering::SeqCst);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Disconnected,
error: None,
reason_code: None,
session_present: None,
});
break;
}
Err(e) => {
connected.store(false, Ordering::SeqCst);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Error,
error: Some(e.to_string()),
reason_code: None,
session_present: None,
});
if !auto_reconnect || first_connect {
engine.log(&id, LogLevel::Error, "connection_error",
format!("连接错误: {e}"), None);
break;
}
// max_reconnect_times>0 时按次数停止重试;0 表示无限重连
if max_reconnect_times > 0 && reconnect_attempts >= max_reconnect_times {
engine.log(&id, LogLevel::Error, "reconnect_give_up",
format!("已重连 {reconnect_attempts} 次仍失败,达到最大重连次数 {max_reconnect_times},停止重连: {e}"), None);
break;
}
reconnect_attempts += 1;
engine.log(&id, LogLevel::Warn, "connection_error",
format!("连接错误,2 秒后重连(第 {reconnect_attempts} 次): {e}"), None);
// 退避等待也必须可取消:否则 close() 的取消信号最长
// 2 秒不可见,期间会对用户已关闭的连接继续重连。
tokio::select! {
_ = cancel_rx.changed() => break,
_ = tokio::time::sleep(Duration::from_secs(2)) => {}
}
}
}
}
_ = cancel_rx.changed() => {
engine.log(&id, LogLevel::Info, "disconnect_cancelled", "连接已取消".into(), None);
break;
}
}
}
// 仅当 map 中仍是本代句柄时才移除,避免误删重连后的新句柄
engine.remove_handle_if(&id, generation);
});
}
+44
View File
@@ -0,0 +1,44 @@
//! 平台专属窗口操作。当前仅 Windows:置顶 / 取消置顶。
/// 窗口置顶 / 取消置顶。返回是否成功(非 Windows 或句柄不可得为 false)。
pub fn set_topmost(window: &gpui_kit::Window, topmost: bool) -> bool {
if !cfg!(target_os = "windows") {
return false;
}
set_topmost_windows(window, topmost)
}
#[cfg(target_os = "windows")]
fn set_topmost_windows(window: &gpui_kit::Window, topmost: bool) -> bool {
use raw_window_handle::RawWindowHandle;
use windows::Win32::Foundation::HWND;
use windows::Win32::UI::WindowsAndMessaging::{
SetWindowPos, HWND_NOTOPMOST, HWND_TOPMOST, SET_WINDOW_POS_FLAGS,
};
// gpui 的 Window 自带同名方法,HasWindowHandle 必须全路径调用
let handle = match raw_window_handle::HasWindowHandle::window_handle(window) {
Ok(h) => h,
Err(_) => return false,
};
// WindowHandle::raw 是私有字段;HasWindowHandle 的实现返回自身拷贝,
// trait 方法在此类型上拿不到 RawWindowHandle,用 deref 后的 as_raw 途径
let raw = handle.as_raw();
#[allow(clippy::infallible_destructuring_match)]
let RawWindowHandle::Win32(w) = raw else {
return false;
};
let hwnd = HWND(w.hwnd.get() as *mut core::ffi::c_void);
let insert = if topmost { HWND_TOPMOST } else { HWND_NOTOPMOST };
let flags = SET_WINDOW_POS_FLAGS(SWP_NOMOVE | SWP_NOSIZE | SWP_NOACTIVATE);
unsafe { SetWindowPos(hwnd, insert, 0, 0, 0, 0, flags).is_ok() }
}
#[cfg(not(target_os = "windows"))]
fn set_topmost_windows(_window: &gpui_kit::Window, _topmost: bool) -> bool {
false
}
const SWP_NOSIZE: u32 = 0x0001;
const SWP_NOMOVE: u32 = 0x0002;
const SWP_NOACTIVATE: u32 = 0x0010;
+3 -46
View File
@@ -1,7 +1,7 @@
//! 进程自身资源采样与窗口置顶辅助。 //! 进程自身资源采样(CPU/内存/线程/句柄)。
//! //!
//! 仅 Windows 实现数据采集(`windows` crate);其他平台采样返回 //! 仅 Windows 实现数据采集(`windows` crate);其他平台采样返回全零数据,
//! 全零数据、置顶返回 false,调用方按缺失能力展示。 //! 调用方按缺失能力展示。窗口置顶等平台操作见 [`crate::platform`]。
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
@@ -180,49 +180,6 @@ pub fn uptime() -> Duration {
START.get().map(|t| t.elapsed()).unwrap_or_default() START.get().map(|t| t.elapsed()).unwrap_or_default()
} }
/// 窗口置顶 / 取消置顶。返回是否成功(非 Windows 或句柄不可得为 false)。
pub fn set_topmost(window: &gpui_kit::Window, topmost: bool) -> bool {
if !cfg!(target_os = "windows") {
return false;
}
set_topmost_windows(window, topmost)
}
#[cfg(target_os = "windows")]
fn set_topmost_windows(window: &gpui_kit::Window, topmost: bool) -> bool {
use raw_window_handle::RawWindowHandle;
use windows::Win32::Foundation::HWND;
use windows::Win32::UI::WindowsAndMessaging::{
SetWindowPos, HWND_NOTOPMOST, HWND_TOPMOST, SET_WINDOW_POS_FLAGS,
};
// gpui 的 Window 自带同名方法,HasWindowHandle 必须全路径调用
let handle = match raw_window_handle::HasWindowHandle::window_handle(window) {
Ok(h) => h,
Err(_) => return false,
};
// WindowHandle::raw 是私有字段;HasWindowHandle 的实现返回自身拷贝,
// trait 方法在此类型上拿不到 RawWindowHandle,用 deref 后的 as_raw 途径
let raw = handle.as_raw();
#[allow(clippy::infallible_destructuring_match)]
let RawWindowHandle::Win32(w) = raw else {
return false;
};
let hwnd = HWND(w.hwnd.get() as *mut core::ffi::c_void);
let insert = if topmost { HWND_TOPMOST } else { HWND_NOTOPMOST };
let flags = SET_WINDOW_POS_FLAGS(SWP_NOMOVE | SWP_NOSIZE | SWP_NOACTIVATE);
unsafe { SetWindowPos(hwnd, insert, 0, 0, 0, 0, flags).is_ok() }
}
#[cfg(not(target_os = "windows"))]
fn set_topmost_windows(_window: &gpui_kit::Window, _topmost: bool) -> bool {
false
}
const SWP_NOSIZE: u32 = 0x0001;
const SWP_NOMOVE: u32 = 0x0002;
const SWP_NOACTIVATE: u32 = 0x0010;
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
-658
View File
@@ -1,658 +0,0 @@
//! OTA 自更新:以 Gitea Release 为更新源。
//!
//! 流程:`check_latest`(比较最新 Release 与当前版本)→ `download_and_verify`
//! (下载到 exe 同目录临时文件并校验 sha256)→ `install_and_restart`
//! (改名自替换后拉起新进程)。全程 HTTPS + sha256 校验。
use std::io::Read;
use std::path::PathBuf;
use sha2::Digest as _;
/// 更新源(自建 Gitea,仓库公开、附件可直接下载,无需认证)。
pub const GITEA_URL: &str = "https://gitea.heavenlybook.cn";
pub const PKG_OWNER: &str = "JoeAllen";
pub const PKG_REPO: &str = "mqttx";
const HTTP_TIMEOUT_SECS: u64 = 30;
/// 发现的可用更新。
#[derive(Debug, Clone)]
pub struct UpdateInfo {
/// 新版本号(不含 v 前缀),如 "1.0.1"
pub version: String,
/// Release 说明(Markdown 原文)
pub notes: String,
/// 附件名(判断 7z 压缩包 / 裸二进制)
pub asset_name: String,
/// 平台对应产物下载地址
pub asset_url: String,
/// sha256 校验文件地址(若 Release 提供了 `.sha256` 侧车)
pub sha256_url: Option<String>,
/// 产物字节数(API 提供,用于下载进度)
pub size: u64,
}
/// 当前版本。测试可用环境变量 `MQTTX_VERSION_OVERRIDE` 伪装旧版本,
/// 验证"发现新版本"路径。
pub fn current_version() -> &'static str {
static VERSION: std::sync::OnceLock<String> = std::sync::OnceLock::new();
VERSION.get_or_init(|| {
std::env::var("MQTTX_VERSION_OVERRIDE")
.unwrap_or_else(|_| env!("CARGO_PKG_VERSION").to_string())
})
}
/// 构建目标的平台三元组,需与 release.yml / publish 脚本的产物命名一致。
pub fn platform_triple() -> &'static str {
if cfg!(windows) {
"x86_64-pc-windows-msvc"
} else if cfg!(target_os = "macos") {
"x86_64-apple-darwin"
} else {
"x86_64-unknown-linux-gnu"
}
}
/// 某版本解压后二进制的暂存名(与 CI 的命名约定 `mqttx-v<ver>-<triple>-mqttx[.exe]` 一致)。
pub fn binary_name_for(version: &str) -> String {
let ext = if cfg!(windows) { ".exe" } else { "" };
format!("mqttx-v{}-{}-mqttx{}", version, platform_triple(), ext)
}
/// 某版本的 7z 压缩附件名(打包端与此约定一致;内部为 [`binary_name_for`])。
pub fn asset_name_for(version: &str) -> String {
format!("{}.7z", binary_name_for(version))
}
fn agent() -> ureq::Agent {
ureq::AgentBuilder::new()
// 整体上限:防止慢速滴流的下载无限占用阻塞线程
.timeout(std::time::Duration::from_secs(10 * 60))
.timeout_connect(std::time::Duration::from_secs(HTTP_TIMEOUT_SECS))
.timeout_read(std::time::Duration::from_secs(HTTP_TIMEOUT_SECS))
.build()
}
fn http_get_json(url: &str) -> Result<serde_json::Value, String> {
let resp = agent()
.get(url)
.call()
.map_err(|e| format!("请求失败: {e}"))?;
resp.into_json::<serde_json::Value>()
.map_err(|e| format!("解析响应失败: {e}"))
}
/// 去掉 v 前缀与预发布后缀(`-` 及之后),返回数字主体。
fn release_part(v: &str) -> &str {
let v = v.trim().trim_start_matches(['v', 'V']);
match v.find('-') {
Some(i) => &v[..i],
None => v,
}
}
/// 从版本字符串提取数字段(容忍 v 前缀;预发布后缀(`-` 之后)不参与比较)。
pub fn parse_version(v: &str) -> Vec<u32> {
release_part(v)
.split('.')
.filter_map(|s| s.parse::<u32>().ok())
.collect()
}
/// a 是否比 b 新(按数字段逐位比较,缺失段视为 0)。
/// 预发布版本(含 `-` 后缀)不会比同数字段的正式版新(semver 语义)。
pub fn is_newer(a: &str, b: &str) -> bool {
let (a, b) = (parse_version(a), parse_version(b));
let len = a.len().max(b.len());
for i in 0..len {
let (x, y) = (a.get(i).copied().unwrap_or(0), b.get(i).copied().unwrap_or(0));
if x != y {
return x > y;
}
}
false
}
/// 从 Release JSON 里挑选当前平台的产物与 sha256 侧车。
/// 优先 7z 压缩包(体积小、下载快);Release 未提供 7z 时回退裸二进制
/// (兼容旧发布方式)。
pub fn pick_assets(rel: &serde_json::Value) -> Option<(String, String, u64, Option<String>)> {
let assets = rel.get("assets")?.as_array()?;
let want_ext = if cfg!(windows) { ".exe" } else { "" };
let triple = platform_triple();
// (后缀过滤器, 说明):第一轮 7z,第二轮裸二进制
for suffix in [".7z", want_ext] {
for a in assets {
let Some(name) = a.get("name").and_then(|n| n.as_str()) else {
continue;
};
if !name.contains(triple)
|| !name.ends_with(suffix)
|| name.ends_with(".sha256")
{
continue;
}
// 7z 轮要排除裸二进制(windows 裸 exe 不以 .7z 结尾,天然互斥)
if suffix == ".7z" && !name.ends_with(&format!("{want_ext}.7z")) && !want_ext.is_empty()
{
continue;
}
let Some(url) = a.get("browser_download_url").and_then(|u| u.as_str()) else {
continue;
};
let url = url.to_string();
let size = a.get("size").and_then(|s| s.as_u64()).unwrap_or(0);
// 侧车命名:<asset>.sha256
let sha = assets
.iter()
.find(|b| {
b.get("name").and_then(|n| n.as_str()) == Some(&format!("{name}.sha256"))
})
.and_then(|b| b.get("browser_download_url"))
.and_then(|u| u.as_str())
.map(|s| s.to_string());
return Some((name.to_string(), url, size, sha));
}
}
None
}
/// 检查更新。返回 Ok(None) 表示已是最新。
pub fn check_latest() -> Result<Option<UpdateInfo>, String> {
let base = format!("{GITEA_URL}/api/v1/repos/{PKG_OWNER}/{PKG_REPO}/releases");
// 优先 latest 端点;旧版 Gitea 没有时回退到列表取第一个非草稿/预发布
let rel = match http_get_json(&format!("{base}/latest")) {
Ok(v) => v,
Err(_) => {
let list = http_get_json(&format!("{base}?limit=5"))?;
let first = list
.as_array()
.and_then(|l| l.iter().find(|r| {
r.get("draft").and_then(|d| d.as_bool()) == Some(false)
&& r.get("prerelease").and_then(|d| d.as_bool()) == Some(false)
}))
.ok_or_else(|| "没有可用 Release".to_string())?;
first.clone()
}
};
let tag = rel
.get("tag_name")
.and_then(|t| t.as_str())
.ok_or_else(|| "Release 缺少 tag_name".to_string())?;
if !is_newer(tag, current_version()) {
return Ok(None);
}
let Some((name, url, size, sha)) = pick_assets(&rel) else {
return Err(format!("Release {tag} 没有当前平台({})的产物", platform_triple()));
};
Ok(Some(UpdateInfo {
version: tag.trim_start_matches('v').to_string(),
notes: rel
.get("body")
.and_then(|b| b.as_str())
.unwrap_or_default()
.to_string(),
asset_name: name,
asset_url: url,
sha256_url: sha,
size,
}))
}
/// 已下载并校验、等待安装的更新包(存放在数据目录的暂存区)。
#[derive(Debug, Clone)]
pub struct StagedUpdate {
pub version: String,
pub path: PathBuf,
}
/// 暂存目录:数据目录下的 `updates/`,重启不丢失。
fn stage_dir() -> Result<PathBuf, String> {
let dir = dirs::data_dir()
.unwrap_or_else(std::env::temp_dir)
.join("MQTTX-GPUI")
.join("updates");
std::fs::create_dir_all(&dir).map_err(|e| format!("创建暂存目录失败: {e}"))?;
Ok(dir)
}
/// 检查更新并静默下载到暂存区(供启动自动检查使用)。
/// 返回 Ok(None) 表示已是最新;错误仅记录、不打扰用户。
pub fn check_and_download() -> Result<Option<StagedUpdate>, String> {
let Some(info) = check_latest()? else {
return Ok(None);
};
// 已暂存同版本则不重复下载
if let Some(staged) = load_staged()
&& staged.version == info.version
{
return Ok(Some(staged));
}
download_and_stage(&info, &|_, _| {}).map(Some)
}
/// 下载并校验到暂存区,返回待安装的更新。
///
/// 附件为 7z 压缩包:下载 → 校验 → 解压出二进制暂存(压缩包随即删除)。
/// 回退路径(裸二进制附件)直接校验暂存。
/// `on_progress(downloaded, total)` 在阻塞线程上回调,total 为 0 表示未知。
pub fn download_and_stage(
info: &UpdateInfo,
on_progress: &dyn Fn(u64, u64),
) -> Result<StagedUpdate, String> {
let dir = stage_dir()?;
let binary_name = binary_name_for(&info.version);
let is_7z = info.asset_name.ends_with(".7z");
let final_name = if is_7z {
asset_name_for(&info.version)
} else {
binary_name.clone()
};
// tmp 名加入毫秒时间戳:同 pid 并发下载不共用文件
let tmp = dir.join(format!(
"{}.{}.{}.tmp",
final_name,
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis())
.unwrap_or(0)
));
let dst = dir.join(&final_name);
if let Err(e) = download_to(info, &tmp, on_progress) {
// 失败路径统一清理临时文件
let _ = std::fs::remove_file(&tmp);
return Err(e);
}
// 清掉其他版本的残留暂存文件,避免堆积与误装旧版
let staged_path = if is_7z {
// 解压到暂存目录,取出二进制后删掉压缩包
if let Err(e) = sevenz_rust::decompress_file(&tmp, &dir) {
let _ = std::fs::remove_file(&tmp);
return Err(format!("解压更新包失败: {e}"));
}
let _ = std::fs::remove_file(&tmp);
let extracted = dir.join(&binary_name);
if !extracted.exists() {
return Err(format!(
"更新包内缺少 {},压缩包内容异常",
binary_name
));
}
// 确认解压产物后再清理其他版本,最后返回二进制路径
remove_other_staged(&dir, &extracted);
extracted
} else {
remove_other_staged(&dir, &dst);
std::fs::rename(&tmp, &dst).map_err(|e| format!("暂存失败: {e}"))?;
dst
};
Ok(StagedUpdate {
version: info.version.clone(),
path: staged_path,
})
}
fn download_to(info: &UpdateInfo, tmp: &std::path::Path, on_progress: &dyn Fn(u64, u64)) -> Result<(), String> {
let resp = agent()
.get(&info.asset_url)
.call()
.map_err(|e| format!("下载失败: {e}"))?;
let mut reader = resp.into_reader();
let total = info.size;
let mut file = std::fs::File::create(tmp).map_err(|e| format!("创建临时文件失败: {e}"))?;
let mut buf = [0u8; 64 * 1024];
let mut downloaded: u64 = 0;
let mut hasher = sha2::Sha256::new();
loop {
let n = reader.read(&mut buf).map_err(|e| format!("下载中断: {e}"))?;
if n == 0 {
break;
}
std::io::Write::write_all(&mut file, &buf[..n])
.map_err(|e| format!("写入临时文件失败: {e}"))?;
hasher.update(&buf[..n]);
downloaded += n as u64;
on_progress(downloaded, total);
}
std::io::Write::flush(&mut file).ok();
drop(file);
// 传输完整性:API 提供了大小却没下满/超量,视为损坏(read 返回 0
// 无法区分干净结束与连接提前断开,必须与预期大小对账)
if total > 0 && downloaded != total {
return Err(format!(
"下载不完整:期望 {total} 字节,实际 {downloaded} 字节"
));
}
// sha256 校验
if let Some(expect) = fetch_expected_sha256(info)? {
let actual = format!("{:x}", hasher.finalize());
if !actual.eq_ignore_ascii_case(&expect) {
return Err(format!(
"校验失败:期望 sha256 {expect},实际 {actual}"
));
}
}
Ok(())
}
/// 删除暂存目录中除 `keep` 外的其他版本暂存文件。
fn remove_other_staged(dir: &std::path::Path, keep: &std::path::Path) {
if let Ok(entries) = std::fs::read_dir(dir) {
for e in entries.flatten() {
let p = e.path();
if p == *keep {
continue;
}
let name = e.file_name();
let name = name.to_string_lossy();
if name.starts_with("mqttx-v")
&& name.contains(platform_triple())
&& !name.ends_with(".tmp")
&& !name.ends_with(".sha256")
&& !name.ends_with(".7z")
{
let _ = std::fs::remove_file(&p);
}
}
}
}
/// 读取 `.sha256` 侧车内容,返回其中的哈希值。没有侧车时返回 None。
fn fetch_expected_sha256(info: &UpdateInfo) -> Result<Option<String>, String> {
let Some(url) = &info.sha256_url else {
return Ok(None);
};
let text = agent()
.get(url)
.call()
.map_err(|e| format!("获取校验文件失败: {e}"))?
.into_string()
.map_err(|e| format!("读取校验文件失败: {e}"))?;
// 常见格式:<hex> <filename> 或纯 hex
let hex = text
.split_whitespace()
.next()
.unwrap_or_default()
.to_lowercase();
if hex.len() == 64 && hex.chars().all(|c| c.is_ascii_hexdigit()) {
Ok(Some(hex))
} else {
Err("校验文件格式异常".into())
}
}
/// 从暂存文件名解析版本号(需匹配当前平台的命名约定;`.tmp` 不算)。
/// 预发布后缀原样保留。
pub fn parse_staged_name(name: &str) -> Option<String> {
let ext = if cfg!(windows) { ".exe" } else { "" };
let suffix = format!("-{}-mqttx{ext}", platform_triple());
if name.contains(".tmp") {
return None;
}
name.strip_prefix("mqttx-v")?
.strip_suffix(&suffix)
.map(|s| s.to_string())
}
/// 扫描暂存区,返回待安装的更新(文件存在且比当前版本新才算)。
/// 多个版本并存时取最新版本(readdir 顺序不保证有序)。
pub fn load_staged() -> Option<StagedUpdate> {
let dir = stage_dir().ok()?;
let mut best: Option<StagedUpdate> = None;
for e in std::fs::read_dir(&dir).ok()?.flatten() {
if !e.path().is_file() {
continue;
}
let name = e.file_name();
let name = name.to_string_lossy().to_string();
let Some(version) = parse_staged_name(&name) else {
continue;
};
if !is_newer(&version, current_version()) {
continue;
}
if best.as_ref().is_none_or(|b| is_newer(&version, &b.version)) {
best = Some(StagedUpdate {
version,
path: e.path(),
});
}
}
best
}
/// 安装暂存的更新:替换自身并拉起新进程(旧文件改名 `.old`,
/// 新进程启动时由 [`cleanup_old`] 清理)。成功后调用方应退出当前进程。
pub fn install_staged(staged: &StagedUpdate) -> Result<(), String> {
let cur = std::env::current_exe().map_err(|e| format!("无法定位自身: {e}"))?;
let mut old_name = cur.clone().into_os_string();
old_name.push(".old");
let old = PathBuf::from(old_name);
// 旧残留先清掉,避免改名失败
let _ = std::fs::remove_file(&old);
std::fs::rename(&cur, &old).map_err(|e| format!("替换失败(可能无写入权限): {e}"))?;
if let Err(e) = std::fs::rename(&staged.path, &cur) {
// 回滚,尽量保住当前可执行文件
let rollback = std::fs::rename(&old, &cur);
return Err(format!(
"替换失败: {e};{}",
rollback_result(rollback, &old)
));
}
if let Err(e) = std::process::Command::new(&cur).spawn() {
// 新文件无法启动(常见:杀毒软件锁定刚落地的文件):恢复旧版本
let _ = std::fs::remove_file(&cur);
let rollback = std::fs::rename(&old, &cur);
return Err(format!(
"启动新版本失败: {e};{}",
rollback_result(rollback, &old)
));
}
clear_install_consent();
Ok(())
}
/// 描述回滚结果;回滚失败时提示 `.old` 备份仍保留在原处。
fn rollback_result(rollback: std::io::Result<()>, old: &std::path::Path) -> String {
match rollback {
Ok(()) => "已恢复原版本".to_string(),
Err(re) => format!(
"回滚失败({re}),原版本备份保留在 {}",
old.display()
),
}
}
/// 安装同意标记:仅当用户在更新对话框明确选择「下次启动安装」后才写入,
/// 启动时据此自动安装——用户未选择的更新不会被强制替换。
fn consent_path() -> Result<PathBuf, String> {
stage_dir().map(|d| d.join("install-consent.txt"))
}
/// 记录对指定版本的安装同意。
pub fn mark_install_consent(version: &str) {
if let Ok(p) = consent_path() {
let _ = std::fs::write(p, version);
}
}
/// 清除安装同意标记。
pub fn clear_install_consent() {
if let Ok(p) = consent_path() {
let _ = std::fs::remove_file(p);
}
}
/// 同意标记是否与暂存的版本匹配(版本不匹配视为未同意)。
pub fn install_consent_matches(staged_version: &str) -> bool {
consent_path()
.ok()
.and_then(|p| std::fs::read_to_string(p).ok())
.map(|v| v.trim() == staged_version)
.unwrap_or(false)
}
/// 用户拒绝本次更新:删除暂存包与同意标记。
pub fn discard_staged(staged: &StagedUpdate) {
let _ = std::fs::remove_file(&staged.path);
clear_install_consent();
}
/// 启动时清理上次更新遗留的旧文件(改名失败的 `.old`、中断的 `.tmp`)。
pub fn cleanup_old() {
if let Ok(cur) = std::env::current_exe()
&& let Some(dir) = cur.parent()
&& let Ok(entries) = std::fs::read_dir(dir)
{
for e in entries.flatten() {
// Windows 文件系统不区分大小写,统一按小写匹配
let name = e.file_name();
let lower = name.to_string_lossy().to_lowercase();
if (lower.ends_with(".old") && lower.starts_with("mqttx"))
|| lower.ends_with(".update.tmp")
{
let _ = std::fs::remove_file(e.path());
}
}
}
// 暂存目录里中断的 .tmp。跳过最近修改的文件,避免误删
// 并行实例正在写入的临时文件。
if let Ok(dir) = stage_dir()
&& let Ok(entries) = std::fs::read_dir(&dir)
{
for e in entries.flatten() {
let name = e.file_name();
if !name.to_string_lossy().ends_with(".tmp") {
continue;
}
let recent = std::fs::metadata(e.path())
.and_then(|m| m.modified())
.ok()
.and_then(|t| t.elapsed().ok())
.is_some_and(|age| age < std::time::Duration::from_secs(10 * 60));
if recent {
continue;
}
let _ = std::fs::remove_file(e.path());
}
}
}
// ─── 单元测试 ────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn version_compare() {
assert!(is_newer("1.0.1", "1.0.0"));
assert!(is_newer("v1.1.0", "1.0.9"));
assert!(is_newer("2.0", "1.9.9"));
// 预发布版本不比同数字段的正式版新(semver 语义)
assert!(!is_newer("1.0.0-beta", "1.0.0"));
assert!(!is_newer("1.0.0-beta.2", "1.0.0"));
assert!(!is_newer("1.0.0", "1.0.0"));
assert!(!is_newer("0.9", "1.0.0"));
// 更高版本号的预发布仍是更新(1.0.1-beta > 1.0.0)
assert!(is_newer("1.0.1-beta", "1.0.0"));
assert_eq!(parse_version("v1.2.3-beta"), vec![1, 2, 3]);
assert_eq!(parse_version("v1.2.3"), vec![1, 2, 3]);
}
#[cfg(windows)]
#[test]
fn staged_name_parsing_keeps_prerelease_suffix() {
let triple = platform_triple();
assert_eq!(
parse_staged_name(&format!("mqttx-v1.0.3-{triple}-mqttx.exe")),
Some("1.0.3".to_string())
);
assert_eq!(
parse_staged_name(&format!("mqttx-v1.0.3-beta.1-{triple}-mqttx.exe")),
Some("1.0.3-beta.1".to_string())
);
assert_eq!(
parse_staged_name(&format!("mqttx-v1.0.3-{triple}-mqttx.exe.1234.56.tmp")),
None
);
assert_eq!(parse_staged_name("其他文件.exe"), None);
}
#[test]
fn asset_naming_matches_ci_convention() {
let ext = if cfg!(windows) { ".exe" } else { "" };
// 附件为 7z 压缩包;解压出的二进制保留原命名
assert_eq!(
asset_name_for("1.0.1"),
format!("mqttx-v1.0.1-{triple}-mqttx{ext}.7z", triple = platform_triple())
);
assert_eq!(
binary_name_for("1.0.1"),
format!("mqttx-v1.0.1-{triple}-mqttx{ext}", triple = platform_triple())
);
}
#[test]
fn pick_assets_prefers_7z_over_raw_binary() {
let rel: serde_json::Value = serde_json::json!({
"tag_name": "v1.0.1",
"assets": [
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe", "size": 21000000,
"browser_download_url": "https://x/win.exe"},
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe.sha256", "size": 65,
"browser_download_url": "https://x/win.exe.sha256"},
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe.7z", "size": 9000000,
"browser_download_url": "https://x/win.7z"},
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe.7z.sha256", "size": 65,
"browser_download_url": "https://x/win.7z.sha256"}
]
});
let (name, url, size, sha) = pick_assets(&rel).expect("应选中 7z");
assert!(name.ends_with(".7z"), "应优先 7z 压缩包,实际 {name}");
assert_eq!(url, "https://x/win.7z");
assert_eq!(size, 9000000);
assert_eq!(sha.as_deref(), Some("https://x/win.7z.sha256"));
}
#[test]
fn pick_assets_falls_back_to_raw_binary() {
// 旧发布方式只有裸二进制:仍可被选中
let rel: serde_json::Value = serde_json::json!({
"tag_name": "v1.0.1",
"assets": [
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe", "size": 19132928,
"browser_download_url": "https://x/win"},
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe.sha256", "size": 65,
"browser_download_url": "https://x/win.sha256"}
]
});
let (name, url, _size, sha) = pick_assets(&rel).expect("应回退裸二进制");
assert!(name.ends_with(".exe"));
assert_eq!(url, "https://x/win");
assert_eq!(sha.as_deref(), Some("https://x/win.sha256"));
}
#[test]
fn pick_assets_skips_sidecars_and_other_platforms() {
let rel: serde_json::Value = serde_json::json!({
"tag_name": "v1.0.1",
"assets": [
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe.sha256", "size": 65,
"browser_download_url": "https://x/side"},
]
});
if cfg!(windows) {
// 只有侧车、没有主体 → 不应选中侧车
assert!(pick_assets(&rel).is_none());
}
}
}
+121
View File
@@ -0,0 +1,121 @@
//! 更新检查:访问 Gitea Release API,比较版本并挑选本平台资产。
use super::*;
pub(super) fn agent() -> ureq::Agent {
ureq::AgentBuilder::new()
// 整体上限:防止慢速滴流的下载无限占用阻塞线程
.timeout(std::time::Duration::from_secs(10 * 60))
.timeout_connect(std::time::Duration::from_secs(HTTP_TIMEOUT_SECS))
.timeout_read(std::time::Duration::from_secs(HTTP_TIMEOUT_SECS))
.build()
}
pub(super) fn http_get_json(url: &str) -> Result<serde_json::Value, String> {
let resp = agent()
.get(url)
.call()
.map_err(|e| format!("请求失败: {e}"))?;
resp.into_json::<serde_json::Value>()
.map_err(|e| format!("解析响应失败: {e}"))
}
pub fn pick_assets(rel: &serde_json::Value) -> Option<(String, String, u64, Option<String>)> {
let assets = rel.get("assets")?.as_array()?;
let want_ext = if cfg!(windows) { ".exe" } else { "" };
let triple = platform_triple();
// (后缀过滤器, 说明):第一轮 7z,第二轮裸二进制
for suffix in [".7z", want_ext] {
for a in assets {
let Some(name) = a.get("name").and_then(|n| n.as_str()) else {
continue;
};
if !name.contains(triple)
|| !name.ends_with(suffix)
|| name.ends_with(".sha256")
{
continue;
}
// 7z 轮要排除裸二进制(windows 裸 exe 不以 .7z 结尾,天然互斥)
if suffix == ".7z" && !name.ends_with(&format!("{want_ext}.7z")) && !want_ext.is_empty()
{
continue;
}
let Some(url) = a.get("browser_download_url").and_then(|u| u.as_str()) else {
continue;
};
let url = url.to_string();
let size = a.get("size").and_then(|s| s.as_u64()).unwrap_or(0);
// 侧车命名:<asset>.sha256
let sha = assets
.iter()
.find(|b| {
b.get("name").and_then(|n| n.as_str()) == Some(&format!("{name}.sha256"))
})
.and_then(|b| b.get("browser_download_url"))
.and_then(|u| u.as_str())
.map(|s| s.to_string());
return Some((name.to_string(), url, size, sha));
}
}
None
}
/// 检查更新。返回 Ok(None) 表示已是最新。
pub fn check_latest() -> Result<Option<UpdateInfo>, String> {
let base = format!("{GITEA_URL}/api/v1/repos/{PKG_OWNER}/{PKG_REPO}/releases");
// 优先 latest 端点;旧版 Gitea 没有时回退到列表取第一个非草稿/预发布
let rel = match http_get_json(&format!("{base}/latest")) {
Ok(v) => v,
Err(_) => {
let list = http_get_json(&format!("{base}?limit=5"))?;
let first = list
.as_array()
.and_then(|l| l.iter().find(|r| {
r.get("draft").and_then(|d| d.as_bool()) == Some(false)
&& r.get("prerelease").and_then(|d| d.as_bool()) == Some(false)
}))
.ok_or_else(|| "没有可用 Release".to_string())?;
first.clone()
}
};
let tag = rel
.get("tag_name")
.and_then(|t| t.as_str())
.ok_or_else(|| "Release 缺少 tag_name".to_string())?;
if !is_newer(tag, current_version()) {
return Ok(None);
}
let Some((name, url, size, sha)) = pick_assets(&rel) else {
return Err(format!("Release {tag} 没有当前平台({})的产物", platform_triple()));
};
Ok(Some(UpdateInfo {
version: tag.trim_start_matches('v').to_string(),
notes: rel
.get("body")
.and_then(|b| b.as_str())
.unwrap_or_default()
.to_string(),
asset_name: name,
asset_url: url,
sha256_url: sha,
size,
}))
}
/// 返回 Ok(None) 表示已是最新;错误仅记录、不打扰用户。
pub fn check_and_download() -> Result<Option<StagedUpdate>, String> {
let Some(info) = check_latest()? else {
return Ok(None);
};
// 已暂存同版本则不重复下载
if let Some(staged) = load_staged()
&& staged.version == info.version
{
return Ok(Some(staged));
}
download_and_stage(&info, &|_, _| {}).map(Some)
}
+319
View File
@@ -0,0 +1,319 @@
//! 更新执行:下载暂存(7z 优先)→ sha256 校验 → 重启自替换安装,以及同意标记与旧档清理。
use std::io::Read;
use std::path::PathBuf;
use super::check::agent;
use super::*;
/// 暂存目录:数据目录下的 `updates/`,重启不丢失。
fn stage_dir() -> Result<PathBuf, String> {
let dir = dirs::data_dir()
.unwrap_or_else(std::env::temp_dir)
.join("MQTTX-GPUI")
.join("updates");
std::fs::create_dir_all(&dir).map_err(|e| format!("创建暂存目录失败: {e}"))?;
Ok(dir)
}
/// 下载并校验到暂存区,返回待安装的更新。
///
/// 附件为 7z 压缩包:下载 → 校验 → 解压出二进制暂存(压缩包随即删除)。
/// 回退路径(裸二进制附件)直接校验暂存。
/// `on_progress(downloaded, total)` 在阻塞线程上回调,total 为 0 表示未知。
pub fn download_and_stage(
info: &UpdateInfo,
on_progress: &dyn Fn(u64, u64),
) -> Result<StagedUpdate, String> {
let dir = stage_dir()?;
let binary_name = binary_name_for(&info.version);
let is_7z = info.asset_name.ends_with(".7z");
let final_name = if is_7z {
asset_name_for(&info.version)
} else {
binary_name.clone()
};
// tmp 名加入毫秒时间戳:同 pid 并发下载不共用文件
let tmp = dir.join(format!(
"{}.{}.{}.tmp",
final_name,
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis())
.unwrap_or(0)
));
let dst = dir.join(&final_name);
if let Err(e) = download_to(info, &tmp, on_progress) {
// 失败路径统一清理临时文件
let _ = std::fs::remove_file(&tmp);
return Err(e);
}
// 清掉其他版本的残留暂存文件,避免堆积与误装旧版
let staged_path = if is_7z {
// 解压到暂存目录,取出二进制后删掉压缩包
if let Err(e) = sevenz_rust::decompress_file(&tmp, &dir) {
let _ = std::fs::remove_file(&tmp);
return Err(format!("解压更新包失败: {e}"));
}
let _ = std::fs::remove_file(&tmp);
let extracted = dir.join(&binary_name);
if !extracted.exists() {
return Err(format!(
"更新包内缺少 {},压缩包内容异常",
binary_name
));
}
// 确认解压产物后再清理其他版本,最后返回二进制路径
remove_other_staged(&dir, &extracted);
extracted
} else {
remove_other_staged(&dir, &dst);
std::fs::rename(&tmp, &dst).map_err(|e| format!("暂存失败: {e}"))?;
dst
};
Ok(StagedUpdate {
version: info.version.clone(),
path: staged_path,
})
}
fn download_to(info: &UpdateInfo, tmp: &std::path::Path, on_progress: &dyn Fn(u64, u64)) -> Result<(), String> {
let resp = agent()
.get(&info.asset_url)
.call()
.map_err(|e| format!("下载失败: {e}"))?;
let mut reader = resp.into_reader();
let total = info.size;
let mut file = std::fs::File::create(tmp).map_err(|e| format!("创建临时文件失败: {e}"))?;
let mut buf = [0u8; 64 * 1024];
let mut downloaded: u64 = 0;
let mut hasher = sha2::Sha256::new();
loop {
let n = reader.read(&mut buf).map_err(|e| format!("下载中断: {e}"))?;
if n == 0 {
break;
}
std::io::Write::write_all(&mut file, &buf[..n])
.map_err(|e| format!("写入临时文件失败: {e}"))?;
hasher.update(&buf[..n]);
downloaded += n as u64;
on_progress(downloaded, total);
}
std::io::Write::flush(&mut file).ok();
drop(file);
// 传输完整性:API 提供了大小却没下满/超量,视为损坏(read 返回 0
// 无法区分干净结束与连接提前断开,必须与预期大小对账)
if total > 0 && downloaded != total {
return Err(format!(
"下载不完整:期望 {total} 字节,实际 {downloaded} 字节"
));
}
// sha256 校验
if let Some(expect) = fetch_expected_sha256(info)? {
let actual = format!("{:x}", hasher.finalize());
if !actual.eq_ignore_ascii_case(&expect) {
return Err(format!(
"校验失败:期望 sha256 {expect},实际 {actual}"
));
}
}
Ok(())
}
/// 删除暂存目录中除 `keep` 外的其他版本暂存文件。
fn remove_other_staged(dir: &std::path::Path, keep: &std::path::Path) {
if let Ok(entries) = std::fs::read_dir(dir) {
for e in entries.flatten() {
let p = e.path();
if p == *keep {
continue;
}
let name = e.file_name();
let name = name.to_string_lossy();
if name.starts_with("mqttx-v")
&& name.contains(platform_triple())
&& !name.ends_with(".tmp")
&& !name.ends_with(".sha256")
&& !name.ends_with(".7z")
{
let _ = std::fs::remove_file(&p);
}
}
}
}
/// 读取 `.sha256` 侧车内容,返回其中的哈希值。没有侧车时返回 None。
fn fetch_expected_sha256(info: &UpdateInfo) -> Result<Option<String>, String> {
let Some(url) = &info.sha256_url else {
return Ok(None);
};
let text = agent()
.get(url)
.call()
.map_err(|e| format!("获取校验文件失败: {e}"))?
.into_string()
.map_err(|e| format!("读取校验文件失败: {e}"))?;
// 常见格式:<hex> <filename> 或纯 hex
let hex = text
.split_whitespace()
.next()
.unwrap_or_default()
.to_lowercase();
if hex.len() == 64 && hex.chars().all(|c| c.is_ascii_hexdigit()) {
Ok(Some(hex))
} else {
Err("校验文件格式异常".into())
}
}
/// 多个版本并存时取最新版本(readdir 顺序不保证有序)。
pub fn load_staged() -> Option<StagedUpdate> {
let dir = stage_dir().ok()?;
let mut best: Option<StagedUpdate> = None;
for e in std::fs::read_dir(&dir).ok()?.flatten() {
if !e.path().is_file() {
continue;
}
let name = e.file_name();
let name = name.to_string_lossy().to_string();
let Some(version) = parse_staged_name(&name) else {
continue;
};
if !is_newer(&version, current_version()) {
continue;
}
if best.as_ref().is_none_or(|b| is_newer(&version, &b.version)) {
best = Some(StagedUpdate {
version,
path: e.path(),
});
}
}
best
}
/// 安装暂存的更新:替换自身并拉起新进程(旧文件改名 `.old`,
/// 新进程启动时由 [`cleanup_old`] 清理)。成功后调用方应退出当前进程。
pub fn install_staged(staged: &StagedUpdate) -> Result<(), String> {
let cur = std::env::current_exe().map_err(|e| format!("无法定位自身: {e}"))?;
let mut old_name = cur.clone().into_os_string();
old_name.push(".old");
let old = PathBuf::from(old_name);
// 旧残留先清掉,避免改名失败
let _ = std::fs::remove_file(&old);
std::fs::rename(&cur, &old).map_err(|e| format!("替换失败(可能无写入权限): {e}"))?;
if let Err(e) = std::fs::rename(&staged.path, &cur) {
// 回滚,尽量保住当前可执行文件
let rollback = std::fs::rename(&old, &cur);
return Err(format!(
"替换失败: {e};{}",
rollback_result(rollback, &old)
));
}
if let Err(e) = std::process::Command::new(&cur).spawn() {
// 新文件无法启动(常见:杀毒软件锁定刚落地的文件):恢复旧版本
let _ = std::fs::remove_file(&cur);
let rollback = std::fs::rename(&old, &cur);
return Err(format!(
"启动新版本失败: {e};{}",
rollback_result(rollback, &old)
));
}
clear_install_consent();
Ok(())
}
/// 描述回滚结果;回滚失败时提示 `.old` 备份仍保留在原处。
fn rollback_result(rollback: std::io::Result<()>, old: &std::path::Path) -> String {
match rollback {
Ok(()) => "已恢复原版本".to_string(),
Err(re) => format!(
"回滚失败({re}),原版本备份保留在 {}",
old.display()
),
}
}
/// 安装同意标记:仅当用户在更新对话框明确选择「下次启动安装」后才写入,
/// 启动时据此自动安装——用户未选择的更新不会被强制替换。
fn consent_path() -> Result<PathBuf, String> {
stage_dir().map(|d| d.join("install-consent.txt"))
}
/// 记录对指定版本的安装同意。
pub fn mark_install_consent(version: &str) {
if let Ok(p) = consent_path() {
let _ = std::fs::write(p, version);
}
}
/// 清除安装同意标记。
pub fn clear_install_consent() {
if let Ok(p) = consent_path() {
let _ = std::fs::remove_file(p);
}
}
/// 同意标记是否与暂存的版本匹配(版本不匹配视为未同意)。
pub fn install_consent_matches(staged_version: &str) -> bool {
consent_path()
.ok()
.and_then(|p| std::fs::read_to_string(p).ok())
.map(|v| v.trim() == staged_version)
.unwrap_or(false)
}
/// 用户拒绝本次更新:删除暂存包与同意标记。
pub fn discard_staged(staged: &StagedUpdate) {
let _ = std::fs::remove_file(&staged.path);
clear_install_consent();
}
/// 启动时清理上次更新遗留的旧文件(改名失败的 `.old`、中断的 `.tmp`)。
pub fn cleanup_old() {
if let Ok(cur) = std::env::current_exe()
&& let Some(dir) = cur.parent()
&& let Ok(entries) = std::fs::read_dir(dir)
{
for e in entries.flatten() {
// Windows 文件系统不区分大小写,统一按小写匹配
let name = e.file_name();
let lower = name.to_string_lossy().to_lowercase();
if (lower.ends_with(".old") && lower.starts_with("mqttx"))
|| lower.ends_with(".update.tmp")
{
let _ = std::fs::remove_file(e.path());
}
}
}
// 暂存目录里中断的 .tmp。跳过最近修改的文件,避免误删
// 并行实例正在写入的临时文件。
if let Ok(dir) = stage_dir()
&& let Ok(entries) = std::fs::read_dir(&dir)
{
for e in entries.flatten() {
let name = e.file_name();
if !name.to_string_lossy().ends_with(".tmp") {
continue;
}
let recent = std::fs::metadata(e.path())
.and_then(|m| m.modified())
.ok()
.and_then(|t| t.elapsed().ok())
.is_some_and(|age| age < std::time::Duration::from_secs(10 * 60));
if recent {
continue;
}
let _ = std::fs::remove_file(e.path());
}
}
}
// ─── 单元测试 ────────────────────────────────────────────────────────────────
+238
View File
@@ -0,0 +1,238 @@
//! OTA 自更新:以 Gitea Release 为更新源。
//!
//! - [`check`] 检查最新版本与资产挑选
//! - [`install`] 下载暂存、校验安装、同意标记与清理
//!
//! 本模块只暴露版本比较/资产命名等纯工具与二级模块的公开 API。
use std::path::PathBuf;
use sha2::Digest as _;
mod check;
mod install;
pub use check::{check_and_download, check_latest, pick_assets};
pub use install::{
cleanup_old, clear_install_consent, discard_staged, download_and_stage,
install_consent_matches, install_staged, load_staged, mark_install_consent,
};
/// 更新源(自建 Gitea,仓库公开、附件可直接下载,无需认证)。
pub const GITEA_URL: &str = "https://gitea.heavenlybook.cn";
pub const PKG_OWNER: &str = "JoeAllen";
pub const PKG_REPO: &str = "mqttx";
const HTTP_TIMEOUT_SECS: u64 = 30;
/// 发现的可用更新。
#[derive(Debug, Clone)]
pub struct UpdateInfo {
/// 新版本号(不含 v 前缀),如 "1.0.1"
pub version: String,
/// Release 说明(Markdown 原文)
pub notes: String,
/// 附件名(判断 7z 压缩包 / 裸二进制)
pub asset_name: String,
/// 平台对应产物下载地址
pub asset_url: String,
/// sha256 校验文件地址(若 Release 提供了 `.sha256` 侧车)
pub sha256_url: Option<String>,
/// 产物字节数(API 提供,用于下载进度)
pub size: u64,
}
/// 当前版本。测试可用环境变量 `MQTTX_VERSION_OVERRIDE` 伪装旧版本,
/// 验证"发现新版本"路径。
pub fn current_version() -> &'static str {
static VERSION: std::sync::OnceLock<String> = std::sync::OnceLock::new();
VERSION.get_or_init(|| {
std::env::var("MQTTX_VERSION_OVERRIDE")
.unwrap_or_else(|_| env!("CARGO_PKG_VERSION").to_string())
})
}
/// 构建目标的平台三元组,需与 release.yml / publish 脚本的产物命名一致。
pub fn platform_triple() -> &'static str {
if cfg!(windows) {
"x86_64-pc-windows-msvc"
} else if cfg!(target_os = "macos") {
"x86_64-apple-darwin"
} else {
"x86_64-unknown-linux-gnu"
}
}
/// 某版本解压后二进制的暂存名(与 CI 的命名约定 `mqttx-v<ver>-<triple>-mqttx[.exe]` 一致)。
pub fn binary_name_for(version: &str) -> String {
let ext = if cfg!(windows) { ".exe" } else { "" };
format!("mqttx-v{}-{}-mqttx{}", version, platform_triple(), ext)
}
/// 某版本的 7z 压缩附件名(打包端与此约定一致;内部为 [`binary_name_for`])。
pub fn asset_name_for(version: &str) -> String {
format!("{}.7z", binary_name_for(version))
}
/// 去掉 v 前缀与预发布后缀(`-` 及之后),返回数字主体。
fn release_part(v: &str) -> &str {
let v = v.trim().trim_start_matches(['v', 'V']);
match v.find('-') {
Some(i) => &v[..i],
None => v,
}
}
/// 从版本字符串提取数字段(容忍 v 前缀;预发布后缀(`-` 之后)不参与比较)。
pub fn parse_version(v: &str) -> Vec<u32> {
release_part(v)
.split('.')
.filter_map(|s| s.parse::<u32>().ok())
.collect()
}
/// a 是否比 b 新(按数字段逐位比较,缺失段视为 0)。
/// 预发布版本(含 `-` 后缀)不会比同数字段的正式版新(semver 语义)。
pub fn is_newer(a: &str, b: &str) -> bool {
let (a, b) = (parse_version(a), parse_version(b));
let len = a.len().max(b.len());
for i in 0..len {
let (x, y) = (a.get(i).copied().unwrap_or(0), b.get(i).copied().unwrap_or(0));
if x != y {
return x > y;
}
}
false
}
/// 从 Release JSON 里挑选当前平台的产物与 sha256 侧车。
/// 优先 7z 压缩包(体积小、下载快);Release 未提供 7z 时回退裸二进制
/// (兼容旧发布方式)。
#[derive(Debug, Clone)]
pub struct StagedUpdate {
pub version: String,
pub path: PathBuf,
}
/// 预发布后缀原样保留。
pub fn parse_staged_name(name: &str) -> Option<String> {
let ext = if cfg!(windows) { ".exe" } else { "" };
let suffix = format!("-{}-mqttx{ext}", platform_triple());
if name.contains(".tmp") {
return None;
}
name.strip_prefix("mqttx-v")?
.strip_suffix(&suffix)
.map(|s| s.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn version_compare() {
assert!(is_newer("1.0.1", "1.0.0"));
assert!(is_newer("v1.1.0", "1.0.9"));
assert!(is_newer("2.0", "1.9.9"));
// 预发布版本不比同数字段的正式版新(semver 语义)
assert!(!is_newer("1.0.0-beta", "1.0.0"));
assert!(!is_newer("1.0.0-beta.2", "1.0.0"));
assert!(!is_newer("1.0.0", "1.0.0"));
assert!(!is_newer("0.9", "1.0.0"));
// 更高版本号的预发布仍是更新(1.0.1-beta > 1.0.0)
assert!(is_newer("1.0.1-beta", "1.0.0"));
assert_eq!(parse_version("v1.2.3-beta"), vec![1, 2, 3]);
assert_eq!(parse_version("v1.2.3"), vec![1, 2, 3]);
}
#[cfg(windows)]
#[test]
fn staged_name_parsing_keeps_prerelease_suffix() {
let triple = platform_triple();
assert_eq!(
parse_staged_name(&format!("mqttx-v1.0.3-{triple}-mqttx.exe")),
Some("1.0.3".to_string())
);
assert_eq!(
parse_staged_name(&format!("mqttx-v1.0.3-beta.1-{triple}-mqttx.exe")),
Some("1.0.3-beta.1".to_string())
);
assert_eq!(
parse_staged_name(&format!("mqttx-v1.0.3-{triple}-mqttx.exe.1234.56.tmp")),
None
);
assert_eq!(parse_staged_name("其他文件.exe"), None);
}
#[test]
fn asset_naming_matches_ci_convention() {
let ext = if cfg!(windows) { ".exe" } else { "" };
// 附件为 7z 压缩包;解压出的二进制保留原命名
assert_eq!(
asset_name_for("1.0.1"),
format!("mqttx-v1.0.1-{triple}-mqttx{ext}.7z", triple = platform_triple())
);
assert_eq!(
binary_name_for("1.0.1"),
format!("mqttx-v1.0.1-{triple}-mqttx{ext}", triple = platform_triple())
);
}
#[test]
fn pick_assets_prefers_7z_over_raw_binary() {
let rel: serde_json::Value = serde_json::json!({
"tag_name": "v1.0.1",
"assets": [
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe", "size": 21000000,
"browser_download_url": "https://x/win.exe"},
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe.sha256", "size": 65,
"browser_download_url": "https://x/win.exe.sha256"},
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe.7z", "size": 9000000,
"browser_download_url": "https://x/win.7z"},
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe.7z.sha256", "size": 65,
"browser_download_url": "https://x/win.7z.sha256"}
]
});
let (name, url, size, sha) = pick_assets(&rel).expect("应选中 7z");
assert!(name.ends_with(".7z"), "应优先 7z 压缩包,实际 {name}");
assert_eq!(url, "https://x/win.7z");
assert_eq!(size, 9000000);
assert_eq!(sha.as_deref(), Some("https://x/win.7z.sha256"));
}
#[test]
fn pick_assets_falls_back_to_raw_binary() {
// 旧发布方式只有裸二进制:仍可被选中
let rel: serde_json::Value = serde_json::json!({
"tag_name": "v1.0.1",
"assets": [
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe", "size": 19132928,
"browser_download_url": "https://x/win"},
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe.sha256", "size": 65,
"browser_download_url": "https://x/win.sha256"}
]
});
let (name, url, _size, sha) = pick_assets(&rel).expect("应回退裸二进制");
assert!(name.ends_with(".exe"));
assert_eq!(url, "https://x/win");
assert_eq!(sha.as_deref(), Some("https://x/win.sha256"));
}
#[test]
fn pick_assets_skips_sidecars_and_other_platforms() {
let rel: serde_json::Value = serde_json::json!({
"tag_name": "v1.0.1",
"assets": [
{"name": "mqttx-v1.0.1-x86_64-pc-windows-msvc-mqttx.exe.sha256", "size": 65,
"browser_download_url": "https://x/side"},
]
});
if cfg!(windows) {
// 只有侧车、没有主体 → 不应选中侧车
assert!(pick_assets(&rel).is_none());
}
}
}