feat: GPUI 版补齐 MQTTX 基础功能对标(分组/SSL证书/订阅增强/消息流控制/设置与日志)

This commit is contained in:
2026-09-28 14:41:15 +08:00
parent bf6937b776
commit c1c4fc011e
10 changed files with 2550 additions and 280 deletions
+141 -1
View File
@@ -94,6 +94,33 @@ pub struct LastWill {
pub response_topic: Option<String>,
}
/// SSL/TLS 自定义配置(CA、双向认证、跳过校验)。
#[derive(Debug, Clone, Default, 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,
@@ -144,6 +171,19 @@ pub struct ConnectionConfig {
/// 创建时间(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,
}
fn default_true() -> bool {
@@ -152,6 +192,14 @@ fn default_true() -> bool {
fn default_keep_alive() -> u16 {
60
}
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 {
@@ -164,7 +212,7 @@ impl ConnectionConfig {
path: "/mqtt".into(),
protocol: ProtocolVersion::V5,
transport,
client_id: format!("mqttx_{}", &uuid::Uuid::new_v4().simple().to_string()[..8]),
client_id: generate_client_id(),
username: String::new(),
password: String::new(),
clean_start: true,
@@ -177,6 +225,10 @@ impl ConnectionConfig {
auto_reconnect: true,
last_will: None,
created_at: chrono::Local::now().timestamp(),
group: None,
connection_timeout_secs: default_connection_timeout(),
max_reconnect_times: 0,
ssl: SslConfig::default(),
}
}
@@ -212,6 +264,24 @@ pub struct Subscription {
/// 本地备注颜色(与 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 {
@@ -222,6 +292,38 @@ impl Subscription {
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,
}
}
}
@@ -655,4 +757,42 @@ mod tests {
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);
}
}
+309 -22
View File
@@ -4,6 +4,7 @@
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use rumqttc::mqttbytes::v5::{
ConnectReturnCode as V5ReturnCode, Packet as V5Packet, Publish as V5Publish,
@@ -12,14 +13,16 @@ use rumqttc::mqttbytes::v5::{
use rumqttc::{
AsyncClient as V5AsyncClient, Broker as V5Broker, ConnectionError as V5Error,
Event as V5Event, EventLoop as V5EventLoop, MqttOptions as V5Options, Outgoing as V5Outgoing,
PublishOptions as V5PublishOptions, QoS as V5QoS, Transport as V5Transport,
PublishOptions as V5PublishOptions, QoS as V5QoS, RetainForwardRule, SubscribeFilterInput,
SubscribeProperties as V5SubscribeProperties, TlsConfiguration, Transport as V5Transport,
};
use rumqttc_v4 as rv4;
use rustls::pki_types::{pem::PemObject, CertificateDer, PrivateKeyDer};
use tokio::sync::watch;
use crate::model::{
ConnectionConfig, ConnectionStatus, Direction, LogEntry, LogLevel, MqttRecord, PublishParams,
Subscription, TransportKind,
SslConfig, SubscribeOptions, Subscription, TransportKind,
};// ─── 引擎事件 ────────────────────────────────────────────────────────────────
#[derive(Debug, Clone)]
@@ -201,6 +204,7 @@ impl MqttEngine {
cancel_rx,
connected,
cfg.auto_reconnect,
cfg.max_reconnect_times,
);
}
EventLoopKind::V4(el) => {
@@ -211,6 +215,7 @@ impl MqttEngine {
cancel_rx,
connected,
cfg.auto_reconnect,
cfg.max_reconnect_times,
);
}
}
@@ -335,7 +340,19 @@ impl MqttEngine {
// ── 订阅 / 取消订阅 ─────────────────────────────────────────────────────
/// 旧签名薄封装:不携带 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 {
@@ -357,8 +374,23 @@ impl MqttEngine {
);
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_tracked(sub.topic.clone(), v5_qos(sub.qos))
.subscribe_many_with_properties_tracked(std::iter::once(filter), properties)
.await
{
Ok(notice) => notice.wait_completion_async().await.map_err(|e| {
@@ -441,20 +473,24 @@ impl MqttEngine {
});
}
/// 测试连接:5 秒内等待 CONNACK,结果通过 channel 返回。
/// 测试连接:使用配置的连接超时(上限 30 秒,防止误填过大值长时间挂起),
/// 结果通过 channel 返回。
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(std::time::Duration::from_secs(5), async {
let result = tokio::time::timeout(Duration::from_secs(secs), async {
test_handshake(&cfg).await
})
.await;
let reply = match result {
Ok(inner) => inner,
Err(_) => Err("连接测试超时(5 秒),请检查主机地址与端口".into()),
Err(_) => Err(format!(
"连接测试超时({secs} 秒),请检查主机地址与端口"
)),
};
let _ = tx.try_send(reply);
});
@@ -483,9 +519,180 @@ enum EventLoopKind {
// ─── 选项构建 ────────────────────────────────────────────────────────────────
/// 连接超时秒数:至少 1 秒,避免误填 0 导致立刻超时。
fn connection_timeout_secs(cfg: &ConnectionConfig) -> u64 {
u64::from(cfg.connection_timeout_secs.max(1))
}
/// SSL 配置非默认时构建自定义 rustls 配置;返回 None 表示沿用 rumqttc 默认 TLS。
/// 文件读取/解析失败在此返回 Err,由 connect 的错误通道上报为连接失败事件。
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`:双向认证。
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。
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 官方示例做法一致)。
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()
}
}
/// 去掉空白字符串,避免表单存了空串也写进报文属性。
fn non_blank(s: Option<String>) -> Option<String> {
s.filter(|v| !v.trim().is_empty())
}
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))
@@ -496,17 +703,40 @@ fn build_v5(
}
TransportKind::Wss => {
let url = format!("wss://{}:{}{}", cfg.host, cfg.port, normalize_path(&cfg.path));
V5Options::try_websocket_with_default_tls(cfg.client_id.clone(), url)
.map_err(|e| format!("{e:?}"))?
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 {
opts.set_transport(
V5Transport::try_tls_with_default_config().map_err(|e| format!("TLS 初始化失败: {e:?}"))?,
);
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 {
@@ -526,12 +756,26 @@ fn build_v5(
}
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,
None,
properties,
);
opts.set_last_will(lw);
}
@@ -544,6 +788,7 @@ fn build_v4(
) -> 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))
@@ -554,15 +799,31 @@ fn build_v4(
}
TransportKind::Wss => {
let url = format!("wss://{}:{}{}", cfg.host, cfg.port, normalize_path(&cfg.path));
V4Options::try_websocket_with_default_tls(cfg.client_id.clone(), url)
.map_err(|e| format!("{e:?}"))?
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 {
opts.set_transport(V4Transport::try_tls_with_default_config().map_err(
|e| format!("TLS 初始化失败: {e:?}"),
)?);
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);
@@ -572,6 +833,7 @@ fn build_v4(
}
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(),
@@ -581,7 +843,12 @@ fn build_v4(
opts.set_last_will(lw);
}
Ok(rv4::AsyncClient::builder(opts).build())
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))
}
fn normalize_path(path: &str) -> String {
@@ -702,10 +969,13 @@ fn spawn_v5_loop(
mut cancel_rx: watch::Receiver<bool>,
connected: Arc<AtomicBool>,
auto_reconnect: bool,
max_reconnect_times: u32,
) {
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() => {
@@ -831,9 +1101,16 @@ fn spawn_v5_loop(
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 秒后重连: {e}"), None);
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
format!("连接错误,2 秒后重连(第 {reconnect_attempts} 次): {e}"), None);
tokio::time::sleep(Duration::from_secs(2)).await;
}
}
}
@@ -854,10 +1131,13 @@ fn spawn_v4_loop(
mut cancel_rx: watch::Receiver<bool>,
connected: Arc<AtomicBool>,
auto_reconnect: bool,
max_reconnect_times: u32,
) {
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() => {
@@ -964,9 +1244,16 @@ fn spawn_v4_loop(
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 秒后重连: {e}"), None);
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
format!("连接错误,2 秒后重连(第 {reconnect_attempts} 次): {e}"), None);
tokio::time::sleep(Duration::from_secs(2)).await;
}
}
}
+102 -3
View File
@@ -49,9 +49,9 @@ impl Storage {
}
let mut text = std::fs::read_to_string(&p)
.with_context(|| format!("读取 {} 失败", p.display()))?;
// 容忍 Windows 编辑器/PowerShell 写入的 UTF-8 BOM
if text.starts_with('\u{feff}') {
text.drain(..3);
// 容忍 Windows 编辑器/PowerShell 写入的 UTF-8 BOM(BOM 只有 1 个字符,逐字符剥离)
if let Some(rest) = text.strip_prefix('\u{feff}') {
text = rest.to_string();
}
if text.trim().is_empty() {
return Ok(None);
@@ -164,3 +164,102 @@ impl Default for Storage {
Self::new()
}
}
// ── 连接导入 / 导出(Phase 2 UI 调用) ──────────────────────────────────────
/// 导出连接为 JSON 数组(与 connections.json 同构:缩进格式化、UTF-8 无 BOM)。
pub fn export_connections_to(path: &Path, conns: &[ConnectionConfig]) -> Result<(), String> {
let text = serde_json::to_string_pretty(conns)
.map_err(|e| format!("序列化连接失败: {e}"))?;
std::fs::write(path, text).map_err(|e| format!("写入 {} 失败: {e}", path.display()))
}
/// 从 JSON 文件导入连接。
///
/// 容错规则:容忍 UTF-8 BOM;顶层允许是数组,也可能是 `{ "connections": [...] }` 对象;
/// 任一项不是合法连接配置时返回带序号的错误信息。
pub fn import_connections_from(path: &Path) -> Result<Vec<ConnectionConfig>, String> {
let bytes = std::fs::read(path).map_err(|e| format!("读取 {} 失败: {e}", path.display()))?;
let mut text =
String::from_utf8(bytes).map_err(|_| format!("{} 不是有效的 UTF-8 文件", path.display()))?;
if let Some(rest) = text.strip_prefix('\u{feff}') {
text = rest.to_string();
}
let value: serde_json::Value =
serde_json::from_str(text.trim()).map_err(|e| format!("解析 JSON 失败: {e}"))?;
let items = match value {
serde_json::Value::Array(items) => items,
serde_json::Value::Object(mut map) => match map.remove("connections") {
Some(serde_json::Value::Array(items)) => items,
_ => {
return Err(
"JSON 对象中缺少 \"connections\" 数组字段".to_string(),
);
}
},
_ => return Err("JSON 顶层应为连接数组".to_string()),
};
let mut conns = Vec::with_capacity(items.len());
for (i, item) in items.into_iter().enumerate() {
match serde_json::from_value::<ConnectionConfig>(item) {
Ok(c) => conns.push(c),
Err(e) => return Err(format!("第 {} 项不是合法的连接配置: {e}", i + 1)),
}
}
Ok(conns)
}
#[cfg(test)]
mod tests {
use super::*;
fn temp_path(name: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"mqttx-store-test-{}-{name}",
uuid::Uuid::new_v4().simple()
))
}
#[test]
fn export_import_roundtrip() {
let path = temp_path("roundtrip.json");
let mut c = ConnectionConfig::new();
c.group = Some("测试组".into());
c.connection_timeout_secs = 25;
c.max_reconnect_times = 5;
export_connections_to(&path, std::slice::from_ref(&c)).expect("导出应成功");
let back = import_connections_from(&path).expect("导入应成功");
assert_eq!(back.len(), 1);
assert_eq!(back[0].id, c.id);
assert_eq!(back[0].group.as_deref(), Some("测试组"));
assert_eq!(back[0].connection_timeout_secs, 25);
assert_eq!(back[0].max_reconnect_times, 5);
let _ = std::fs::remove_file(&path);
}
#[test]
fn import_tolerates_bom_and_object_form() {
let path = temp_path("bom-object.json");
let c = ConnectionConfig::new();
let body = serde_json::to_string(&serde_json::json!({
"connections": [serde_json::to_value(&c).unwrap()]
}))
.unwrap();
// 写入 BOM 前缀,验证容错
std::fs::write(&path, format!("\u{feff}{body}")).expect("写入测试文件");
let back = import_connections_from(&path).expect("BOM + 对象形式应可导入");
assert_eq!(back.len(), 1);
let _ = std::fs::remove_file(&path);
}
#[test]
fn import_reports_invalid_item_index() {
let path = temp_path("invalid.json");
let good = serde_json::to_value(ConnectionConfig::new()).unwrap();
let body = serde_json::json!([good, {"id": "bad"}]).to_string();
std::fs::write(&path, body).expect("写入测试文件");
let err = import_connections_from(&path).expect_err("非法项应报错");
assert!(err.contains("第 2 项"), "错误应定位到序号: {err}");
let _ = std::fs::remove_file(&path);
}
}
+392 -9
View File
@@ -1,6 +1,7 @@
//! 应用外壳:标题栏 + 连接侧边栏 + 多标签连接视图 + 引擎事件泵。
use std::collections::{HashMap, VecDeque};
use std::path::PathBuf;
use gpui_kit::component::button::{Button, ButtonVariants as _};
use gpui_kit::component::scroll::ScrollableElement as _;
@@ -13,14 +14,16 @@ use gpui_kit::component::{
};
use gpui_kit::prelude::FluentBuilder as _;
use gpui_kit::{
div, px, App, AppContext as _, Context, Entity, InteractiveElement as _, StatefulInteractiveElement as _,
IntoElement, ParentElement as _, Render, SharedString, Styled as _, Window,
div, px, App, AppContext as _, Context, Entity, InteractiveElement as _, IntoElement,
KeyBinding, ParentElement as _, Render, SharedString, StatefulInteractiveElement as _,
Styled as _, Window,
};
use crate::aliyun::AliyunPreset;
use crate::model::{
AppSettings, ConnectionConfig, ConnectionStatus, Direction, GlobalVariable, LogEntry,
LogLevel, MqttRecord, PublishParams, PublishPreset, Subscription, ThemeModePref,
LogLevel, MqttRecord, PublishParams, PublishPreset, SubscribeOptions, Subscription,
ThemeModePref,
};
use crate::mqtt::{EngineEvent, MqttEngine};
use crate::store::Storage;
@@ -33,6 +36,35 @@ use crate::ui::{
const MAX_LOGS: usize = 3000;
// 应用级快捷键动作(context=None:任意焦点状态下都匹配)。
gpui_kit::actions!(mqttx, [NewConnection, OpenSettings, ExportConnections, ImportConnections]);
/// 侧边栏分组过滤;chips 只做列表过滤,分组重命名/删除通过编辑连接的分组字段完成。
#[derive(Clone, PartialEq, Eq)]
enum GroupFilter {
All,
Ungrouped,
Named(String),
}
/// 全局 action 回调只有 `&mut App`:先 defer 出当前分发栈,再取活动窗口回到实体执行,
/// 避免在窗口借用期间重入 `update`。
fn run_on_active_window(
cx: &mut App,
weak: gpui_kit::WeakEntity<MqttXApp>,
f: impl FnOnce(Entity<MqttXApp>, &mut Window, &mut App) + 'static,
) {
let Some(handle) = cx.active_window() else {
return;
};
cx.defer(move |cx| {
let Some(entity) = weak.upgrade() else {
return;
};
let _ = handle.update(cx, move |_, window, cx| f(entity, window, cx));
});
}
pub struct MqttXApp {
storage: Storage,
pub engine: std::sync::Arc<MqttEngine>,
@@ -59,6 +91,10 @@ pub struct MqttXApp {
views: HashMap<String, Entity<ConnectionView>>,
search: Entity<InputState>,
/// 分组过滤状态(chips 点击切换,与搜索条件叠加)
group_filter: GroupFilter,
/// 原生文件对话框进行中标记(对话框阻塞后台线程,防止重复弹出)
file_dialog_open: bool,
}
impl MqttXApp {
@@ -88,7 +124,51 @@ impl MqttXApp {
let search = cx.new(|cx| InputState::new(window, cx).placeholder("搜索连接…"));
// ── 全局快捷键:绑定 + action 注册 ──
// 元素级 on_action 依赖焦点路径,无焦点时分发从树根开始收不到;
// 全局 action 在冒泡末端必达,经 defer 取窗口避免在分发栈内重入。
cx.bind_keys([
KeyBinding::new("ctrl-n", NewConnection, None),
KeyBinding::new("ctrl-,", OpenSettings, None),
KeyBinding::new("ctrl-shift-e", ExportConnections, None),
KeyBinding::new("ctrl-shift-i", ImportConnections, None),
]);
let weak = cx.entity().downgrade();
// Context 上有同名 on_action(绘制期注册),这里必须走 App 的全局注册
App::on_action::<NewConnection>(cx, move |_, cx| {
let w = weak.clone();
run_on_active_window(cx, w, |entity, window, cx| {
if window.has_active_dialog(cx) {
return;
}
entity.update(cx, |app, cx| app.open_connection_form(None, window, cx));
});
});
let weak = cx.entity().downgrade();
App::on_action::<OpenSettings>(cx, move |_, cx| {
let w = weak.clone();
run_on_active_window(cx, w, |entity, window, cx| {
if window.has_active_dialog(cx) {
return;
}
crate::ui::settings_dialog::open(entity, window, cx);
});
});
let weak = cx.entity().downgrade();
App::on_action::<ExportConnections>(cx, move |_, cx| {
let w = weak.clone();
run_on_active_window(cx, w, |entity, window, cx| {
entity.update(cx, |app, cx| app.export_connections(window, cx));
});
});
let weak = cx.entity().downgrade();
App::on_action::<ImportConnections>(cx, move |_, cx| {
let w = weak.clone();
run_on_active_window(cx, w, |entity, window, cx| {
entity.update(cx, |app, cx| app.import_connections(window, cx));
});
});
Self {
storage,
engine,
@@ -108,6 +188,8 @@ impl MqttXApp {
active_tab: None,
views: HashMap::new(),
search,
group_filter: GroupFilter::All,
file_dialog_open: false,
}
}
@@ -206,7 +288,7 @@ impl MqttXApp {
}
fn on_connected(&mut self, connection_id: &str) {
// 连接成功后自动恢复已保存订阅(auto resubscribe)。
// 连接成功后自动恢复已保存订阅(auto resubscribe),禁用的订阅跳过。
let Some(config) = self.connections.iter().find(|c| c.id == connection_id).cloned() else {
return;
};
@@ -217,10 +299,13 @@ impl MqttXApp {
.subscriptions
.iter()
.filter(|s| s.connection_id == connection_id)
.filter(|s| s.enabled)
.cloned()
.collect();
for sub in subs {
self.engine.subscribe(connection_id.to_string(), sub);
let opts = SubscribeOptions::from(&sub);
self.engine
.subscribe_with_options(connection_id.to_string(), sub, opts);
}
}
@@ -242,12 +327,61 @@ impl MqttXApp {
}
fn push_log(&mut self, entry: LogEntry) {
// 落盘只在事件路径执行(不在 render 热路径),与内存环形缓冲互不影响
self.write_log_file(&entry);
if self.logs.len() >= MAX_LOGS {
self.logs.pop_front();
}
self.logs.push_back(entry);
}
/// 追加写入按日切分的日志文件 `mqttx-YYYY-MM-DD.log`。
/// 每条都尝试写入;失败只在首次 eprintln 一次后保持静默,且绝不回调
/// log/push_log(防止递归触发日志)。
fn write_log_file(&self, entry: &LogEntry) {
use std::io::Write as _;
use std::sync::atomic::{AtomicBool, Ordering};
static WRITE_FAILED: AtomicBool = AtomicBool::new(false);
let result = (|| -> std::io::Result<()> {
use chrono::TimeZone as _;
let ts = chrono::Local
.timestamp_millis_opt(entry.timestamp)
.single()
.unwrap_or_else(chrono::Local::now);
let dir = self.storage.log_dir();
std::fs::create_dir_all(&dir)?;
let path = dir.join(format!("mqttx-{}.log", ts.format("%Y-%m-%d")));
// 连接名查不到时回退为 id,保证行格式恒定
let conn = self
.connections
.iter()
.find(|c| c.id == entry.connection_id)
.map(|c| c.name.as_str())
.unwrap_or(entry.connection_id.as_str());
let mut file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(path)?;
writeln!(
file,
"{} [{}] {}/{}: {}",
ts.format("%Y-%m-%d %H:%M:%S%.3f"),
entry.level.label(),
conn,
entry.event,
entry.message
)
})();
if let Err(e) = result
&& !WRITE_FAILED.swap(true, Ordering::Relaxed)
{
eprintln!("[app] 写入日志文件失败(后续静默): {e}");
}
}
pub fn log(&mut self, conn: &str, level: LogLevel, event: &str, message: String) {
self.push_log(LogEntry {
timestamp: chrono::Local::now().timestamp_millis(),
@@ -386,6 +520,134 @@ impl MqttXApp {
self.storage.save_settings(&self.settings);
}
/// 数据目录(设置对话框展示 / 打开用)。
pub fn data_dir(&self) -> PathBuf {
self.storage.dir().to_path_buf()
}
/// 日志目录(设置对话框展示 / 打开用)。
pub fn log_dir(&self) -> PathBuf {
self.storage.log_dir()
}
// ── 连接导入 / 导出 ────────────────────────────────────────────────────
/// 在后台线程弹出阻塞式原生文件对话框,完成后回主线程执行 `done`。
/// 对话框进行中置位,防止重复弹出。
fn spawn_file_dialog<R, F, D>(
&mut self,
work: F,
done: D,
window: &mut Window,
cx: &mut Context<Self>,
) where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
D: FnOnce(&mut Self, R, &mut Window, &mut Context<Self>) + 'static,
{
if self.file_dialog_open {
return;
}
self.file_dialog_open = true;
// rfd 的阻塞对话框不可占用 UI 线程:结果经 smol 通道送回异步任务
let (tx, rx) = smol::channel::bounded::<R>(1);
std::thread::spawn(move || {
let _ = tx.send_blocking(work());
});
let weak = cx.entity().downgrade();
cx.spawn_in(window, async move |_this, cx: &mut gpui_kit::AsyncWindowContext| {
let result = rx.recv().await.ok();
weak.update_in(cx, |app, window, cx| {
app.file_dialog_open = false;
if let Some(result) = result {
done(app, result, window, cx);
}
})
.ok();
})
.detach();
}
/// 导出全部连接为 JSON(原生保存对话框,默认文件名 connections-export.json)。
pub fn export_connections(&mut self, window: &mut Window, cx: &mut Context<Self>) {
let dir = self.storage.dir().to_path_buf();
self.spawn_file_dialog(
move || {
rfd::FileDialog::new()
.set_title("导出连接")
.set_directory(dir)
.set_file_name("connections-export.json")
.add_filter("JSON", &["json"])
.save_file()
},
|app, path, window, cx| {
let Some(path) = path else {
return; // 用户取消
};
let count = app.connections.len();
match crate::store::export_connections_to(&path, &app.connections) {
Ok(()) => window.push_notification(
Notification::success(format!(
"已导出 {count} 条连接到 {}",
path.display()
)),
cx,
),
Err(e) => window.push_notification(Notification::error(e), cx),
}
},
window,
cx,
);
}
/// 导入连接:按 id 去重(同 id 跳过),新 id 保留文件原值;读取解析在后台线程完成。
pub fn import_connections(&mut self, window: &mut Window, cx: &mut Context<Self>) {
let dir = self.storage.dir().to_path_buf();
self.spawn_file_dialog(
move || {
let Some(path) = rfd::FileDialog::new()
.set_title("导入连接")
.set_directory(dir)
.add_filter("JSON", &["json"])
.pick_file()
else {
return Ok(None); // 用户取消
};
crate::store::import_connections_from(&path).map(Some)
},
|app, result, window, cx| match result {
Err(e) => {
window.push_notification(Notification::error(e), cx);
}
Ok(None) => {}
Ok(Some(items)) => {
let existing: std::collections::HashSet<String> =
app.connections.iter().map(|c| c.id.clone()).collect();
let mut added = 0usize;
let mut skipped = 0usize;
for item in items {
if existing.contains(&item.id) {
skipped += 1;
} else {
app.connections.push(item);
added += 1;
}
}
if added > 0 {
app.storage.save_connections(&app.connections);
}
window.push_notification(
Notification::success(format!("导入 {added} 条,跳过 {skipped} 条")),
cx,
);
}
},
window,
cx,
);
}
// ── 对话框 ────────────────────────────────────────────────────────────
pub fn open_connection_form(
@@ -396,7 +658,16 @@ impl MqttXApp {
) {
let app = cx.entity().downgrade();
let engine = self.engine.clone();
let form = cx.new(|cx| ConnectionForm::new(engine, edit, window, cx));
// 表单的分组下拉需要现有分组列表(去重、排序)
let mut groups: Vec<String> = self
.connections
.iter()
.filter_map(|c| c.group.as_ref().map(|g| g.trim().to_string()))
.filter(|g| !g.is_empty())
.collect();
groups.sort();
groups.dedup();
let form = cx.new(|cx| ConnectionForm::new(engine, edit, groups, window, cx));
window.open_dialog(cx, move |dialog, _, cx| {
let form_ok = form.clone();
let form_test = form.clone();
@@ -550,6 +821,8 @@ impl MqttXApp {
let w1 = app_weak.clone();
let w2 = app_weak.clone();
let w3 = app_weak.clone();
let w4 = app_weak.clone();
let w5 = app_weak.clone();
menu.item(
PopupMenuItem::new("阿里云设备")
.on_click(move |_, window, cx| {
@@ -566,6 +839,25 @@ impl MqttXApp {
},
))
.separator()
.item(PopupMenuItem::new("导出连接").on_click(
move |_, window, cx| {
if let Some(entity) = w4.upgrade() {
entity.update(cx, |app, cx| {
app.export_connections(window, cx)
});
}
},
))
.item(PopupMenuItem::new("导入连接").on_click(
move |_, window, cx| {
if let Some(entity) = w5.upgrade() {
entity.update(cx, |app, cx| {
app.import_connections(window, cx)
});
}
},
))
.separator()
.item(PopupMenuItem::new("设置").on_click(
move |_, window, cx| {
if let Some(entity) = w3.upgrade() {
@@ -602,12 +894,23 @@ impl MqttXApp {
fn render_sidebar(&mut self, _window: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
let query = self.search.read(cx).value().to_lowercase();
let search = self.search.clone();
let filter = self.group_filter.clone();
let total = self.connections.len();
let hover_bg = cx.theme().muted;
let mut rows = v_flex().gap_0p5().flex_1().overflow_y_scrollbar();
let mut shown = 0usize;
for conn in self.connections.clone() {
// 分组过滤与搜索叠加(AND)
let group = conn.group.as_deref().map(str::trim).unwrap_or("");
let match_group = match &filter {
GroupFilter::All => true,
GroupFilter::Ungrouped => group.is_empty(),
GroupFilter::Named(name) => group == name.as_str(),
};
if !match_group {
continue;
}
if !query.is_empty() && !conn.name.to_lowercase().contains(&query) {
continue;
}
@@ -692,8 +995,6 @@ impl MqttXApp {
.text_color(cx.theme().muted_foreground)
.child("还没有连接,点击上方 + 新建"),
)
} else if query.is_empty() {
v_flex().flex_1().child(rows)
} else if shown == 0 {
v_flex()
.flex_1()
@@ -748,9 +1049,91 @@ impl MqttXApp {
.child(v_flex().px_2().pb_2().child(Input::new(&search).small().prefix(
gpui_kit::component::Icon::new(IconName::Search).small(),
)))
.child(self.render_group_chips(cx))
.child(body)
}
/// 搜索框下方的分组 chips:全部 / 未分组 / 各分组(带计数徽标),点击切换过滤。
fn render_group_chips(&self, cx: &mut Context<Self>) -> impl IntoElement {
let mut names: Vec<String> = self
.connections
.iter()
.filter_map(|c| c.group.as_deref().map(str::trim))
.filter(|g| !g.is_empty())
.map(str::to_string)
.collect();
names.sort();
names.dedup();
let ungrouped = self
.connections
.iter()
.filter(|c| c.group.as_deref().map(str::trim).unwrap_or("").is_empty())
.count();
let mut row = h_flex().gap_1().flex_wrap().px_2().pb_1();
row = row.child(self.render_group_chip(
"chip-all",
format!("全部 {}", self.connections.len()),
GroupFilter::All,
cx,
));
row = row.child(self.render_group_chip(
"chip-ungrouped",
format!("未分组 {ungrouped}"),
GroupFilter::Ungrouped,
cx,
));
for name in names {
let count = self
.connections
.iter()
.filter(|c| c.group.as_deref().map(str::trim) == Some(name.as_str()))
.count();
row = row.child(self.render_group_chip(
SharedString::from(format!("chip-{}", name)),
format!("{name} {count}"),
GroupFilter::Named(name),
cx,
));
}
row
}
fn render_group_chip(
&self,
id: impl Into<gpui_kit::ElementId>,
label: String,
filter: GroupFilter,
cx: &mut Context<Self>,
) -> impl IntoElement {
let selected = match (&self.group_filter, &filter) {
(GroupFilter::All, GroupFilter::All)
| (GroupFilter::Ungrouped, GroupFilter::Ungrouped) => true,
(GroupFilter::Named(a), GroupFilter::Named(b)) => a == b,
_ => false,
};
div()
.id(id)
.px_2()
.py_0p5()
.rounded_md()
.text_xs()
.when(selected, |d| d.bg(cx.theme().secondary).font_semibold())
.when(!selected, |d| {
d.bg(cx.theme().muted)
.text_color(cx.theme().muted_foreground)
.hover(|d| d.bg(cx.theme().secondary))
})
.cursor_pointer()
.on_click(cx.listener(move |this, _, _, cx| {
// listener 是 Fn(可多次调用),只能克隆而不能移出捕获的 filter
this.group_filter = filter.clone();
cx.notify();
}))
.child(label)
}
fn render_conn_power(
&self,
cfg: &ConnectionConfig,
+176 -4
View File
@@ -3,6 +3,7 @@
use std::sync::Arc;
use gpui_kit::component::scroll::ScrollableElement as _;
use gpui_kit::component::button::{Button, ButtonVariants as _};
use gpui_kit::component::input::{Input, InputState};
use gpui_kit::component::select::{Select, SelectEvent, SelectState};
use gpui_kit::component::switch::Switch;
@@ -18,18 +19,22 @@ use gpui_kit::{
Styled as _, Subscription, Window,
};
use crate::model::{ConnectionConfig, LastWill, ProtocolVersion, TransportKind};
use crate::model::{ConnectionConfig, LastWill, ProtocolVersion, SslConfig, TransportKind};
use crate::mqtt::MqttEngine;
use crate::ui::widgets::{field, make_select, OptionDelegate};
const PROTOCOLS: [&str; 2] = ["MQTT 5.0", "MQTT 3.1.1"];
const TRANSPORTS: [&str; 4] = ["TCP", "TLS", "WebSocket", "WSS"];
const QOS: [&str; 3] = ["QoS 0", "QoS 1", "QoS 2"];
/// 分组下拉的首项:选择它表示不归属任何分组。
const NO_GROUP: &str = "无分组";
pub struct ConnectionForm {
engine: Arc<MqttEngine>,
editing_id: Option<String>,
created_at: i64,
/// 已有连接的分组列表(去重),用于分组下拉
groups: Vec<String>,
name: Entity<InputState>,
host: Entity<InputState>,
@@ -43,24 +48,39 @@ pub struct ConnectionForm {
receive_max: Entity<InputState>,
max_packet: Entity<InputState>,
topic_alias: Entity<InputState>,
conn_timeout: Entity<InputState>,
max_reconnect: Entity<InputState>,
will_topic: Entity<InputState>,
will_payload: Entity<InputState>,
will_content_type: Entity<InputState>,
will_response_topic: Entity<InputState>,
// SSL/TLS 区块(transport 为 TLS/WSS 时显示)
ssl_ca: Entity<InputState>,
ssl_client_cert: Entity<InputState>,
ssl_client_key: Entity<InputState>,
// 新分组名输入;非空时优先于下拉选择
group_new: Entity<InputState>,
protocol: Entity<SelectState<OptionDelegate>>,
transport: Entity<SelectState<OptionDelegate>>,
will_qos: Entity<SelectState<OptionDelegate>>,
group_select: Entity<SelectState<OptionDelegate>>,
clean_start: bool,
auto_resubscribe: bool,
auto_reconnect: bool,
will_enabled: bool,
will_retain: bool,
ssl_ignore_ca: bool,
// 按需显示:高级分组默认折叠
show_auth: bool,
show_mqtt5: bool,
show_will: bool,
show_ssl: bool,
testing: bool,
_subs: Vec<Subscription>,
@@ -105,6 +125,7 @@ impl ConnectionForm {
pub fn new(
engine: Arc<MqttEngine>,
edit: Option<ConnectionConfig>,
groups: Vec<String>,
window: &mut Window,
cx: &mut Context<Self>,
) -> Self {
@@ -127,6 +148,18 @@ impl ConnectionForm {
.masked(true)
});
let keep_alive = input(&c.keep_alive.to_string(), "60", window, cx);
let conn_timeout = input(
&c.connection_timeout_secs.to_string(),
"秒(默认 10)",
window,
cx,
);
let max_reconnect = input(
&c.max_reconnect_times.to_string(),
"0=无限重连",
window,
cx,
);
let session_expiry = input(
&if c.session_expiry_interval > 0 {
c.session_expiry_interval.to_string()
@@ -141,7 +174,13 @@ impl ConnectionForm {
let max_packet = optional_input(c.maximum_packet_size, "最大报文长度", window, cx);
let topic_alias = optional_input(c.topic_alias_maximum.map(u32::from), "主题别名上限", window, cx);
let (will_topic, will_payload, will_enabled, will_retain, will_qos_idx) =
let ssl_ca = input(&c.ssl.ca_file, "CA 证书 PEM 路径(可选)", window, cx);
let ssl_client_cert = input(&c.ssl.client_cert_file, "客户端证书 PEM 路径(可选)", window, cx);
let ssl_client_key = input(&c.ssl.client_key_file, "客户端私钥 PEM 路径(可选)", window, cx);
let group_new = input("", "新分组名(可选)", window, cx);
let (will_topic, will_payload, will_enabled, will_retain, will_qos_idx, will_content_type, will_response_topic) =
match &c.last_will {
Some(w) => (
input(&w.topic, "遗嘱主题", window, cx),
@@ -149,6 +188,8 @@ impl ConnectionForm {
true,
w.retain,
w.qos as usize,
input(w.content_type.as_deref().unwrap_or(""), "内容类型(可选)", window, cx),
input(w.response_topic.as_deref().unwrap_or(""), "响应主题(可选)", window, cx),
),
None => (
input("", "遗嘱主题", window, cx),
@@ -156,6 +197,8 @@ impl ConnectionForm {
false,
false,
0,
input("", "内容类型(可选)", window, cx),
input("", "响应主题(可选)", window, cx),
),
};
@@ -171,6 +214,16 @@ impl ConnectionForm {
.unwrap_or(0);
let transport = make_select(&TRANSPORTS, transport_idx, window, cx);
let will_qos = make_select(&QOS, will_qos_idx, window, cx);
// 分组下拉:首项固定为「无分组」,其后是已有分组
let mut group_labels = vec![NO_GROUP];
group_labels.extend(groups.iter().map(|s| s.as_str()));
let group_idx = c
.group
.as_deref()
.and_then(|g| groups.iter().position(|x| x == g))
.map(|i| i + 1)
.unwrap_or(0);
let group_select = make_select(&group_labels, group_idx, window, cx);
// 切换传输方式时,把端口同步成该方式的默认端口,并刷新 WS 路径的按需显示。
let port_for_sub = port.clone();
@@ -203,6 +256,7 @@ impl ConnectionForm {
engine,
editing_id,
created_at,
groups,
name,
host,
port,
@@ -211,23 +265,34 @@ impl ConnectionForm {
username,
password,
keep_alive,
conn_timeout,
max_reconnect,
session_expiry,
receive_max,
max_packet,
topic_alias,
ssl_ca,
ssl_client_cert,
ssl_client_key,
group_new,
will_topic,
will_payload,
will_content_type,
will_response_topic,
protocol,
transport,
will_qos,
group_select,
clean_start: c.clean_start,
auto_resubscribe: c.auto_resubscribe,
auto_reconnect: c.auto_reconnect,
will_enabled,
will_retain,
ssl_ignore_ca: c.ssl.ignore_ca,
show_auth: true,
show_mqtt5: false,
show_will: false,
show_ssl: false,
testing: false,
_subs: vec![sub_transport, sub_protocol],
}
@@ -298,6 +363,48 @@ impl ConnectionForm {
.map(|v| u16::try_from(v).map_err(|_| "主题别名上限不能超过 65535".to_string()))
.transpose()?;
let conn_timeout_raw = self.val(&self.conn_timeout, cx);
let connection_timeout_secs = if conn_timeout_raw.trim().is_empty() {
10
} else {
conn_timeout_raw
.trim()
.parse::<u16>()
.ok()
.filter(|v| *v >= 1)
.ok_or_else(|| "连接超时需要是 1~65535 的整数(秒)".to_string())?
};
let max_reconnect_raw = self.val(&self.max_reconnect, cx);
let max_reconnect_times = if max_reconnect_raw.trim().is_empty() {
0
} else {
max_reconnect_raw
.trim()
.parse::<u32>()
.map_err(|_| "最大重连次数需要是非负整数(0=无限)".to_string())?
};
// 分组:新分组名非空时优先,否则取下拉所选(首项=无分组)
let group = {
let new_name = self.val(&self.group_new, cx);
let trimmed = new_name.trim();
if !trimmed.is_empty() {
Some(trimmed.to_string())
} else {
let idx = self
.group_select
.read(cx)
.selected_value()
.copied()
.unwrap_or(0);
if idx == 0 {
None
} else {
self.groups.get(idx - 1).cloned()
}
}
};
let protocol = match self.protocol.read(cx).selected_value() {
Some(1) => ProtocolVersion::V311,
_ => ProtocolVersion::V5,
@@ -309,18 +416,35 @@ impl ConnectionForm {
.and_then(|i| TransportKind::ALL.get(*i).copied())
.unwrap_or_default();
let ssl = SslConfig {
ca_file: self.val(&self.ssl_ca, cx).trim().to_string(),
client_cert_file: self.val(&self.ssl_client_cert, cx).trim().to_string(),
client_key_file: self.val(&self.ssl_client_key, cx).trim().to_string(),
ignore_ca: self.ssl_ignore_ca,
};
// 只填证书或只填私钥必然握手失败;仅 TLS/WSS 下 SSL 生效,按生效传输校验
if transport.is_tls()
&& ssl.client_cert_file.is_empty() != ssl.client_key_file.is_empty()
{
return Err("客户端证书与客户端密钥必须同时填写".into());
}
let last_will = if self.will_enabled {
let topic = self.val(&self.will_topic, cx);
if topic.trim().is_empty() {
return Err("遗嘱主题不能为空".into());
}
let opt_str = |v: String| {
let t = v.trim().to_string();
if t.is_empty() { None } else { Some(t) }
};
Some(LastWill {
topic,
payload: self.val(&self.will_payload, cx),
qos: self.will_qos.read(cx).selected_value().copied().unwrap_or(0) as u8,
retain: self.will_retain,
content_type: None,
response_topic: None,
content_type: opt_str(self.val(&self.will_content_type, cx)),
response_topic: opt_str(self.val(&self.will_response_topic, cx)),
})
} else {
None
@@ -348,6 +472,10 @@ impl ConnectionForm {
auto_reconnect: self.auto_reconnect,
last_will,
created_at: self.created_at,
group,
connection_timeout_secs,
max_reconnect_times,
ssl,
})
}
@@ -406,6 +534,7 @@ impl ConnectionForm {
"auto_reconnect" => this.auto_reconnect = v,
"will_enabled" => this.will_enabled = v,
"will_retain" => this.will_retain = v,
"ssl_ignore_ca" => this.ssl_ignore_ca = v,
_ => {}
}
cx.notify();
@@ -490,6 +619,11 @@ impl Render for ConnectionForm {
self.transport.read(cx).selected_value(),
Some(2) | Some(3)
);
// SSL/TLS 区块只对 TLS/WSS 有意义
let is_tls = matches!(
self.transport.read(cx).selected_value(),
Some(1) | Some(3)
);
v_flex()
.gap_3()
.max_h(px(560.))
@@ -508,7 +642,24 @@ impl Render for ConnectionForm {
.child(div().flex_1().min_w(px(0.)).child(field("传输", Select::new(&self.transport))))
)
.child(h_flex().gap_2().w_full()
.child(div().flex_1().min_w(px(0.)).child(field("分组", Select::new(&self.group_select))))
.child(div().flex_1().min_w(px(0.)).child(field("新建分组(优先于左侧选择)", Input::new(&self.group_new))))
)
.child(h_flex().gap_2().w_full().items_end()
.child(div().flex_1().min_w(px(0.)).child(field("Client ID", Input::new(&self.client_id))))
// 一键换一个随机 Client ID,与 ConnectionConfig::new 的生成逻辑一致
.child(
Button::new("cid-regen")
.icon(IconName::RefreshCw)
.ghost()
.tooltip("重新生成 Client ID")
.on_click(cx.listener(|this, _, window, cx| {
let cid = crate::model::generate_client_id();
this.client_id
.update(cx, |s, cx| s.set_value(cid, window, cx));
cx.notify();
})),
)
// WS 路径仅对 WebSocket/WSS 有意义,按需显示
.when(is_ws, |w| w.child(
div().flex_1().min_w(px(0.)).child(field("WS 路径", Input::new(&self.path)))
@@ -525,11 +676,27 @@ impl Render for ConnectionForm {
.child(div().flex_1().min_w(px(0.)).child(field("密码", Input::new(&self.password).mask_toggle())))
.child(div().w(px(110.)).child(field("Keep Alive (秒)", Input::new(&self.keep_alive))))
)
.child(h_flex().gap_2().w_full()
.child(div().w(px(150.)).child(field("连接超时 (秒)", Input::new(&self.conn_timeout))))
.child(div().w(px(170.)).child(field("最大重连次数 (0=无限)", Input::new(&self.max_reconnect))))
)
.child(self.switch_row("clean_start", "Clean Start / Clean Session", self.clean_start, cx))
.child(self.switch_row("auto_reconnect", "断线自动重连", self.auto_reconnect, cx))
.child(self.switch_row("auto_resubscribe", "连接后自动恢复订阅", self.auto_resubscribe, cx)),
cx,
))
// ── SSL/TLS(仅 TLS/WSS 显示) ──
.when(is_tls, |w| w.child(self.collapsible_section(
"sec-ssl", "SSL/TLS", None, false, self.show_ssl,
|f| f.show_ssl = !f.show_ssl, border, card_bg,
v_flex().gap_2()
// Phase 2 或后续接原生文件对话框;GPUI 暂无内置文件选择器,先用路径文本输入
.child(field("CA 证书 (PEM)", Input::new(&self.ssl_ca)))
.child(field("客户端证书 (PEM)", Input::new(&self.ssl_client_cert)))
.child(field("客户端密钥 (PEM)", Input::new(&self.ssl_client_key)))
.child(self.switch_row("ssl_ignore_ca", "忽略 CA 校验(不验证服务器证书)", self.ssl_ignore_ca, cx)),
cx,
)))
// ── MQTT 5 高级属性(仅 v5 协议显示) ──
.when(is_v5, |w| w.child(self.collapsible_section(
"sec-mqtt5", "MQTT 5 属性", None, false, self.show_mqtt5,
@@ -554,6 +721,11 @@ impl Render for ConnectionForm {
.child(div().w(px(130.)).child(field("QoS", Select::new(&self.will_qos))))
)
.child(field("遗嘱内容", Input::new(&self.will_payload)))
// v5 遗嘱属性;3.1.1 连接会忽略这两项
.child(h_flex().gap_2().w_full()
.child(div().flex_1().min_w(px(0.)).child(field("内容类型 (Content Type)", Input::new(&self.will_content_type))))
.child(div().flex_1().min_w(px(0.)).child(field("响应主题 (Response Topic)", Input::new(&self.will_response_topic))))
)
.child(self.switch_row("will_retain", "Retain", self.will_retain, cx))
),
cx,
File diff suppressed because it is too large Load Diff
+170 -9
View File
@@ -1,11 +1,15 @@
//! 应用设置对话框:主题、消息缓存条数、时间戳显示。
//! 应用设置对话框:主题、消息缓存、时间戳、自动检查更新、数据/日志目录、关于。
use std::path::{Path, PathBuf};
use gpui_kit::component::button::{Button, ButtonVariants as _};
use gpui_kit::component::clipboard::Clipboard;
use gpui_kit::component::input::{Input, InputState};
use gpui_kit::component::select::{Select, SelectState};
use gpui_kit::component::switch::Switch;
use gpui_kit::component::{
h_flex, notification::Notification, v_flex, WindowExt as _,
h_flex, notification::Notification, v_flex, ActiveTheme as _, Sizable as _, StyledExt as _,
WindowExt as _,
};
use gpui_kit::{
div, px, App, AppContext as _, Context, Entity, IntoElement, ParentElement as _, Render,
@@ -15,17 +19,30 @@ use gpui_kit::{
use crate::model::{AppSettings, ThemeModePref};
use crate::ui::app::MqttXApp;
use crate::ui::widgets::{field, make_select, OptionDelegate};
use crate::ui::IconName;
const THEMES: [&str; 3] = ["跟随系统", "浅色", "深色"];
const SITE_URL: &str = "https://mqttx.app";
const REPO_URL: &str = "https://gitea.heavenlybook.cn/JoeAllen/mqttx";
struct SettingsDialog {
theme: Entity<SelectState<OptionDelegate>>,
max_messages: Entity<InputState>,
show_millis: bool,
auto_check_update: bool,
data_dir: PathBuf,
log_dir: PathBuf,
}
impl SettingsDialog {
fn new(initial: AppSettings, window: &mut Window, cx: &mut Context<Self>) -> Self {
fn new(
initial: AppSettings,
data_dir: PathBuf,
log_dir: PathBuf,
window: &mut Window,
cx: &mut Context<Self>,
) -> Self {
let theme_idx = match initial.theme {
ThemeModePref::System => 0,
ThemeModePref::Light => 1,
@@ -37,6 +54,9 @@ impl SettingsDialog {
InputState::new(window, cx).default_value(initial.max_messages.to_string())
}),
show_millis: initial.show_millis,
auto_check_update: initial.auto_check_update,
data_dir,
log_dir,
}
}
@@ -54,20 +74,41 @@ impl SettingsDialog {
.parse::<usize>()
.map_err(|_| "消息缓存条数必须是正整数".to_string())?
.max(100);
// 四个字段全部显式写出,避免 ..Default::default() 把开关项重置为默认值
Ok(AppSettings {
theme,
max_messages,
show_millis: self.show_millis,
..Default::default()
auto_check_update: self.auto_check_update,
})
}
}
/// 用系统文件管理器打开目录;失败弹 toast(不写日志,避免与日志落盘互相触发)。
fn open_in_file_manager(path: &Path, window: &mut Window, cx: &mut App) {
#[cfg(target_os = "windows")]
let mut cmd = std::process::Command::new("explorer");
#[cfg(target_os = "macos")]
let mut cmd = std::process::Command::new("open");
#[cfg(all(unix, not(target_os = "macos")))]
let mut cmd = std::process::Command::new("xdg-open");
cmd.arg(path);
if let Err(e) = cmd.spawn() {
window.push_notification(Notification::error(format!("打开目录失败: {e}")), cx);
}
}
impl Render for SettingsDialog {
fn render(&mut self, _window: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
let data_dir = self.data_dir.to_string_lossy().into_owned();
let log_dir = self.log_dir.to_string_lossy().into_owned();
let data_dir_for_open = self.data_dir.clone();
let log_dir_for_open = self.log_dir.clone();
v_flex()
.gap_4()
.w(px(420.))
.w(px(500.))
.child(field("主题", Select::new(&self.theme)))
.child(field(
"每条连接内存中保留的消息条数",
@@ -86,20 +127,140 @@ impl Render for SettingsDialog {
})),
),
)
.child(
h_flex()
.justify_between()
.child(div().text_sm().child("自动检查更新"))
.child(
Switch::new("auto-check-update")
.checked(self.auto_check_update)
.on_change(cx.listener(|this, v, _, cx| {
this.auto_check_update = *v;
cx.notify();
})),
),
)
.child(div().h(px(1.)).bg(cx.theme().border).w_full())
.child(field(
"数据目录",
h_flex()
.gap_1()
.w_full()
.child(
div()
.flex_1()
.min_w(px(0.))
.text_xs()
.whitespace_nowrap()
.overflow_hidden()
.text_ellipsis()
.text_color(cx.theme().muted_foreground)
.child(data_dir.clone()),
)
.child(Clipboard::new("copy-data-dir").value(data_dir.clone()))
.child(
Button::new("open-data-dir")
.icon(IconName::FolderOpen)
.label("打开目录")
.outline()
.xsmall()
.on_click(move |_, window, cx| {
open_in_file_manager(&data_dir_for_open, window, cx);
}),
),
))
.child(field(
"日志目录",
h_flex()
.gap_1()
.w_full()
.child(
div()
.flex_1()
.min_w(px(0.))
.text_xs()
.whitespace_nowrap()
.overflow_hidden()
.text_ellipsis()
.text_color(cx.theme().muted_foreground)
.child(log_dir.clone()),
)
.child(
Button::new("open-log-dir")
.icon(IconName::FolderOpen)
.label("打开日志目录")
.outline()
.xsmall()
.on_click(move |_, window, cx| {
// 日志目录可能尚未创建(从未产生过日志)
if let Err(e) = std::fs::create_dir_all(&log_dir_for_open) {
window.push_notification(
Notification::error(format!("创建日志目录失败: {e}")),
cx,
);
return;
}
open_in_file_manager(&log_dir_for_open, window, cx);
}),
),
))
.child(div().h(px(1.)).bg(cx.theme().border).w_full())
.child(
v_flex().gap_1().child(
v_flex()
.gap_1()
.child(div().text_sm().font_semibold().child("关于"))
.child(
h_flex()
.gap_2()
.text_xs()
.text_color(cx.theme().muted_foreground)
.child(format!("版本 {}", env!("CARGO_PKG_VERSION")))
.child("Apache-2.0 许可证"),
)
.child(
div()
.text_xs()
.text_color(cx.theme().muted_foreground)
.child("对标 MQTTX 的 GPUI 跨平台 MQTT 调试客户端"),
)
.child(
h_flex().gap_1().child(
Button::new("about-site")
.label("mqttx.app")
.ghost()
.xsmall()
.tooltip(SITE_URL)
.on_click(|_, _, cx| cx.open_url(SITE_URL)),
).child(
Button::new("about-repo")
.label("gitea.heavenlybook.cn/JoeAllen/mqttx")
.ghost()
.xsmall()
.tooltip(REPO_URL)
.on_click(|_, _, cx| cx.open_url(REPO_URL)),
),
),
),
)
}
}
pub fn open(app: Entity<MqttXApp>, window: &mut Window, cx: &mut App) {
let initial = app.read(cx).settings.clone();
let dialog_view: Entity<SettingsDialog> =
cx.new(|cx| SettingsDialog::new(initial, window, cx));
let (initial, data_dir, log_dir) = {
let state = app.read(cx);
(state.settings.clone(), state.data_dir(), state.log_dir())
};
let dialog_view: Entity<SettingsDialog> = cx.new(|cx| {
SettingsDialog::new(initial, data_dir, log_dir, window, cx)
});
let app_save = app.clone();
window.open_dialog(cx, move |dialog, _window, _cx| {
let body = dialog_view.clone();
let for_collect = dialog_view.clone();
let app_for_save = app_save.clone();
dialog
.w(px(480.))
.w(px(560.))
.title("设置")
.child(body)
.footer(