独立监控核心和模块
This commit is contained in:
zhongluofeng
@@ -0,0 +1,470 @@
|
||||
// monitor_kernel.rs — ThingHK 硬件监控内核的 Tauri 侧集成
|
||||
//
|
||||
// 职责（见 ThingHK_GUIDE.md 第 2.1 节 + 阶段三）：
|
||||
// 1. 准备 Kernel 可执行文件（从 binaries/ 资源目录复制到工作目录）
|
||||
// 2. 构造 StartProcessParams 交由 ProcessManager 拉起/重启（不自己管生命周期）
|
||||
// 3. 作为 HTTP 客户端：轮询 /status 判断就绪 → 订阅 /stream SSE → emit "monitor-data"
|
||||
// 4. 写入熔断：SSE 断开后停止转发，监听 process-status-changed 在 Kernel 恢复后重新订阅
|
||||
//
|
||||
// 数据流：Kernel --SSE--> MonitorKernel(本文件) --emit--> 前端 MonitorModule.vue
|
||||
//
|
||||
// 注意：Kernel 的 stdout/stderr 被 ProcessManager 设为 null，所有数据交互走 HTTP。
|
||||
|
||||
use std::fs;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use futures_util::StreamExt;
|
||||
use reqwest::Client;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tauri::path::BaseDirectory;
|
||||
use tauri::{AppHandle, Emitter, Listener, Manager};
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
use crate::process_manager::{ProcessManager, StartProcessParams};
|
||||
|
||||
// ===================== 常量 =====================
|
||||
|
||||
/// Kernel 进程在 ProcessManager 中的 id（与 monitor 模块 index.ts 的 process.name 对应）
|
||||
const PROCESS_ID: &str = "monitor";
|
||||
/// Kernel 监听端口（与 ThingHK 默认端口一致，见 ThingHK Program.cs DefaultPort）
|
||||
const KERNEL_PORT: u16 = 8730;
|
||||
/// 冷启动就绪轮询间隔（与 mihomo 经验一致）
|
||||
const READY_POLL_INTERVAL_MS: u64 = 500;
|
||||
/// 冷启动就绪总超时（阶段一实测冷启动约 5s，留 5s 余量）
|
||||
const READY_TIMEOUT_MS: u64 = 10_000;
|
||||
|
||||
// ===================== 数据结构 =====================
|
||||
|
||||
/// Kernel /status 响应（与 ThingHK Contracts.cs KernelStatus 对应）
|
||||
#[derive(Serialize, Deserialize, Clone, Debug)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct KernelStatus {
|
||||
pub ready: bool,
|
||||
pub is_admin: bool,
|
||||
pub uptime_ms: f64,
|
||||
pub group_count: u32,
|
||||
pub sensor_count: u32,
|
||||
pub providers: Vec<String>,
|
||||
pub schema_version: u32,
|
||||
}
|
||||
|
||||
/// Kernel /snapshot 与 /stream 推送的传感器快照（与 ThingHK Contracts.cs SensorSnapshot 对应）
|
||||
/// 这里用 serde_json::Value 透传，避免 Rust 侧重复定义完整 schema：
|
||||
/// Kernel 的 schemaVersion=1 契约由 Kernel 维护，前端按 schemaVersion 解析。
|
||||
pub type SensorSnapshot = serde_json::Value;
|
||||
|
||||
/// 返回给前端的 Kernel 信息
|
||||
#[derive(Serialize, Clone)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MonitorKernelInfo {
|
||||
pub path: String,
|
||||
pub exists: bool,
|
||||
pub port: u16,
|
||||
}
|
||||
|
||||
/// 返回给前端的运行状态
|
||||
#[derive(Serialize, Clone)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MonitorStatus {
|
||||
pub running: bool,
|
||||
pub pid: Option<u32>,
|
||||
pub ready: bool,
|
||||
pub sensor_count: u32,
|
||||
pub restart_count: u32,
|
||||
}
|
||||
|
||||
// ===================== MonitorKernel =====================
|
||||
|
||||
/// Tauri 侧的 Kernel 客户端。
|
||||
/// 持有 HTTP client 和订阅控制句柄，不持有进程句柄（进程由 ProcessManager 管理）。
|
||||
/// 实现 Clone：sub_handle / listener_ids 用 Arc<Mutex> 共享，
|
||||
/// 这样从 tauri::State clone 出的实例与原实例共享订阅控制状态。
|
||||
#[derive(Clone)]
|
||||
pub struct MonitorKernel {
|
||||
root: PathBuf,
|
||||
client: Client,
|
||||
/// 无超时 client，专用于 SSE 长连接（/stream）
|
||||
sse_client: Client,
|
||||
/// SSE 订阅任务句柄，用于在 stop 时取消订阅
|
||||
sub_handle: Arc<Mutex<Option<tauri::async_runtime::JoinHandle<()>>>>,
|
||||
/// 监听 process-status-changed 的句柄，用于在 stop 时取消监听
|
||||
listener_ids: Arc<Mutex<Vec<tauri::EventId>>>,
|
||||
}
|
||||
|
||||
impl MonitorKernel {
|
||||
pub fn new(app_data_dir: PathBuf) -> Self {
|
||||
let root = app_data_dir.join("monitor");
|
||||
fs::create_dir_all(&root).ok();
|
||||
Self {
|
||||
root,
|
||||
client: Client::builder()
|
||||
.timeout(Duration::from_secs(30))
|
||||
.build()
|
||||
.unwrap_or_else(|_| Client::new()),
|
||||
sse_client: Client::builder()
|
||||
.build()
|
||||
.unwrap_or_else(|_| Client::new()),
|
||||
sub_handle: Arc::new(Mutex::new(None)),
|
||||
listener_ids: Arc::new(Mutex::new(Vec::new())),
|
||||
}
|
||||
}
|
||||
|
||||
fn cores_dir(&self) -> PathBuf {
|
||||
self.root.join("cores")
|
||||
}
|
||||
pub fn kernel_path(&self) -> PathBuf {
|
||||
self.cores_dir().join("ThingHK.exe")
|
||||
}
|
||||
fn kernel_url(&self) -> String {
|
||||
format!("http://127.0.0.1:{}", KERNEL_PORT)
|
||||
}
|
||||
|
||||
/// 确保内核就位：若 cores/ 无内核，从资源目录复制
|
||||
pub fn prepare_kernel(&self, app: &AppHandle) -> Result<MonitorKernelInfo, String> {
|
||||
let kernel = self.kernel_path();
|
||||
if !kernel.exists() {
|
||||
if let Ok(res) = app.path().resolve("binaries/ThingHK.exe", BaseDirectory::Resource) {
|
||||
if res.exists() {
|
||||
fs::create_dir_all(self.cores_dir()).ok();
|
||||
fs::copy(&res, &kernel).map_err(|e| format!("复制 Kernel 失败: {}", e))?;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(MonitorKernelInfo {
|
||||
path: kernel.to_string_lossy().to_string(),
|
||||
exists: kernel.exists(),
|
||||
port: KERNEL_PORT,
|
||||
})
|
||||
}
|
||||
|
||||
/// 构造启动 Kernel 的进程参数（交由 ProcessManager.start 拉起）
|
||||
pub fn prepare_for_start(&self, app: &AppHandle) -> Result<StartProcessParams, String> {
|
||||
let info = self.prepare_kernel(app)?;
|
||||
if !info.exists {
|
||||
return Err(format!(
|
||||
"ThingHK Kernel 未安装。请将 ThingHK.exe 放置到 src-tauri/binaries/ 后重新构建，或直接放到：\n{}",
|
||||
self.cores_dir().to_string_lossy()
|
||||
));
|
||||
}
|
||||
Ok(StartProcessParams {
|
||||
id: PROCESS_ID.into(),
|
||||
executable: self.kernel_path().to_string_lossy().to_string(),
|
||||
args: vec![
|
||||
"serve".into(),
|
||||
"--port".into(),
|
||||
KERNEL_PORT.to_string(),
|
||||
"--basic".into(),
|
||||
],
|
||||
cwd: Some(self.cores_dir().to_string_lossy().to_string()),
|
||||
name: "ThingHK".into(),
|
||||
restart_on_crash: true,
|
||||
max_restarts: 3,
|
||||
})
|
||||
}
|
||||
|
||||
/// 查询 Kernel /status（不启动订阅）
|
||||
pub async fn get_status(&self) -> Result<KernelStatus, String> {
|
||||
let url = format!("{}/status", self.kernel_url());
|
||||
let resp = self
|
||||
.client
|
||||
.get(&url)
|
||||
.timeout(Duration::from_secs(3))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| format!("请求 Kernel /status 失败: {}", e))?;
|
||||
if !resp.status().is_success() {
|
||||
return Err(format!("Kernel /status 返回 HTTP {}", resp.status()));
|
||||
}
|
||||
resp.json().await.map_err(|e| format!("解析 Kernel /status 失败: {}", e))
|
||||
}
|
||||
|
||||
/// 一次性拉取 /snapshot
|
||||
pub async fn get_snapshot(&self) -> Result<SensorSnapshot, String> {
|
||||
let url = format!("{}/snapshot", self.kernel_url());
|
||||
let resp = self
|
||||
.client
|
||||
.get(&url)
|
||||
.timeout(Duration::from_secs(5))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| format!("请求 Kernel /snapshot 失败: {}", e))?;
|
||||
if !resp.status().is_success() {
|
||||
return Err(format!("Kernel /snapshot 返回 HTTP {}", resp.status()));
|
||||
}
|
||||
resp.json().await.map_err(|e| format!("解析 Kernel /snapshot 失败: {}", e))
|
||||
}
|
||||
|
||||
/// 启动 SSE 订阅循环。
|
||||
/// 流程：轮询 /status 等 ready → 订阅 /stream → 解析 SSE 事件 → emit "monitor-data"
|
||||
/// 写入熔断：SSE 断开后停止 emit，等待外部调用 reconnect 或 process-status-changed 触发重连
|
||||
pub async fn start_subscription(self: Self, app: AppHandle) {
|
||||
// 1. 轮询等待 Kernel ready（冷启动约 5s）
|
||||
if let Err(e) = self.wait_for_ready(&app).await {
|
||||
eprintln!("[monitor] 等待 Kernel ready 失败，订阅不启动: {}", e);
|
||||
let _ = app.emit("monitor-error", serde_json::json!({ "stage": "ready", "message": e }));
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. 订阅 SSE
|
||||
self.run_sse_loop(app).await;
|
||||
}
|
||||
|
||||
/// 轮询 /status 直到 ready 或超时
|
||||
async fn wait_for_ready(&self, app: &AppHandle) -> Result<(), String> {
|
||||
let url = format!("{}/status", self.kernel_url());
|
||||
let deadline = std::time::Instant::now() + Duration::from_millis(READY_TIMEOUT_MS);
|
||||
let mut last_err = String::new();
|
||||
while std::time::Instant::now() < deadline {
|
||||
match self
|
||||
.client
|
||||
.get(&url)
|
||||
.timeout(Duration::from_secs(2))
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
Ok(resp) if resp.status().is_success() => {
|
||||
match resp.json::<KernelStatus>().await {
|
||||
Ok(s) if s.ready => {
|
||||
let _ = app.emit(
|
||||
"monitor-ready",
|
||||
serde_json::json!({
|
||||
"isAdmin": s.is_admin,
|
||||
"sensorCount": s.sensor_count,
|
||||
"providers": s.providers,
|
||||
}),
|
||||
);
|
||||
return Ok(());
|
||||
}
|
||||
Ok(_) => {} // 还没 ready，继续轮询
|
||||
Err(e) => last_err = e.to_string(),
|
||||
}
|
||||
}
|
||||
Ok(resp) => last_err = format!("HTTP {}", resp.status()),
|
||||
Err(e) => last_err = e.to_string(),
|
||||
}
|
||||
// 通知前端正在加载（前端可显示 "Kernel 启动中..."）
|
||||
let elapsed = READY_TIMEOUT_MS.saturating_sub(deadline.duration_since(std::time::Instant::now()).as_millis() as u64);
|
||||
let _ = app.emit("monitor-loading", serde_json::json!({ "elapsedMs": elapsed }));
|
||||
tokio::time::sleep(Duration::from_millis(READY_POLL_INTERVAL_MS)).await;
|
||||
}
|
||||
Err(format!("Kernel 在 {}ms 内未就绪: {}", READY_TIMEOUT_MS, last_err))
|
||||
}
|
||||
|
||||
/// SSE 订阅主循环。
|
||||
/// 断开后自动重试（带退避），实现写入熔断 + 自动重连。
|
||||
async fn run_sse_loop(self: Self, app: AppHandle) {
|
||||
let url = format!("{}/stream", self.kernel_url());
|
||||
loop {
|
||||
match self.subscribe_once(&url, &app).await {
|
||||
// 正常结束（客户端取消或服务端关闭）
|
||||
Ok(()) => {
|
||||
eprintln!("[monitor] SSE 流正常结束");
|
||||
break;
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("[monitor] SSE 流异常断开: {}，3s 后重试", e);
|
||||
let _ = app.emit(
|
||||
"monitor-disconnected",
|
||||
serde_json::json!({ "message": e }),
|
||||
);
|
||||
tokio::time::sleep(Duration::from_secs(3)).await;
|
||||
// 重连前先确认 Kernel 是否还活着（可能已被 stop）
|
||||
if !self.is_kernel_alive().await {
|
||||
eprintln!("[monitor] Kernel 已停止，退出 SSE 循环");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 订阅一次 SSE 流，直到断开。
|
||||
/// 解析 `event: snapshot\ndata: {json}\n\n` 格式，emit "monitor-data"。
|
||||
async fn subscribe_once(&self, url: &str, app: &AppHandle) -> Result<(), String> {
|
||||
let resp = self
|
||||
.sse_client
|
||||
.get(url)
|
||||
.header("Accept", "text/event-stream")
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| format!("请求 /stream 失败: {}", e))?;
|
||||
|
||||
if !resp.status().is_success() {
|
||||
return Err(format!("/stream 返回 HTTP {}", resp.status()));
|
||||
}
|
||||
|
||||
let mut stream = resp.bytes_stream();
|
||||
let mut buffer = String::new();
|
||||
|
||||
while let Some(chunk) = stream.next().await {
|
||||
let chunk = chunk.map_err(|e| format!("读取 SSE chunk 失败: {}", e))?;
|
||||
// SSE 是文本协议，按 UTF-8 解码追加到缓冲区
|
||||
buffer.push_str(&String::from_utf8_lossy(&chunk));
|
||||
|
||||
// 按双换行分割事件（SSE 事件以空行分隔）
|
||||
while let Some(pos) = buffer.find("\n\n") {
|
||||
let event_str = buffer[..pos].to_string();
|
||||
buffer.drain(..pos + 2);
|
||||
|
||||
if let Some(json_str) = parse_sse_data(&event_str) {
|
||||
if let Ok(snap) = serde_json::from_str::<SensorSnapshot>(&json_str) {
|
||||
let _ = app.emit("monitor-data", snap);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 检查 Kernel 是否还在响应（用于 SSE 断开后判断是否应重连）
|
||||
async fn is_kernel_alive(&self) -> bool {
|
||||
let url = format!("{}/status", self.kernel_url());
|
||||
self.client
|
||||
.get(&url)
|
||||
.timeout(Duration::from_secs(2))
|
||||
.send()
|
||||
.await
|
||||
.map(|r| r.status().is_success())
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
/// 停止 SSE 订阅（进程由 ProcessManager.stop 负责）
|
||||
pub async fn stop_subscription(&self, app: &AppHandle) {
|
||||
// 取消 SSE 任务
|
||||
if let Some(handle) = self.sub_handle.lock().await.take() {
|
||||
handle.abort();
|
||||
}
|
||||
// 取消 process-status-changed 监听
|
||||
let ids = self.listener_ids.lock().await.drain(..).collect::<Vec<_>>();
|
||||
for id in ids {
|
||||
app.unlisten(id);
|
||||
}
|
||||
}
|
||||
|
||||
/// 注册 process-status-changed 监听：当 Kernel 进程被自动重启恢复 Running 时，
|
||||
/// 自动重新启动 SSE 订阅（实现崩溃恢复后的自愈）。
|
||||
pub async fn register_auto_reconnect(self: Self, app: AppHandle) {
|
||||
let app_clone = app.clone();
|
||||
let this = self.clone();
|
||||
let id = app.listen("process-status-changed", move |event| {
|
||||
// 只关心 monitor 进程的状态变化
|
||||
// ProcessInfo 只有 Serialize，这里用 Value 解析
|
||||
if let Ok(v) = serde_json::from_str::<serde_json::Value>(event.payload()) {
|
||||
if v.get("id").and_then(|i| i.as_str()) == Some(PROCESS_ID) {
|
||||
let is_running = v.get("status").and_then(|s| s.as_str()) == Some("running");
|
||||
if is_running {
|
||||
let this = this.clone();
|
||||
let app = app_clone.clone();
|
||||
tauri::async_runtime::spawn(async move {
|
||||
eprintln!("[monitor] 检测到 Kernel 重启恢复，重新订阅 SSE");
|
||||
// 重启后需要重新等待 ready（冷启动约 5s）
|
||||
this.clone().start_subscription(app).await;
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
self.listener_ids.lock().await.push(id);
|
||||
}
|
||||
|
||||
/// 启动 Kernel 进程并开始 SSE 订阅（命令和 setup 自动启动共用）。
|
||||
/// ProcessManager 通过 app.state 获取，无需外部传入。
|
||||
pub async fn start_with_subscription(
|
||||
&self,
|
||||
app: &AppHandle,
|
||||
) -> Result<crate::process_manager::ProcessInfo, String> {
|
||||
let pm = app.state::<ProcessManager>();
|
||||
let params = self.prepare_for_start(app)?;
|
||||
let info = pm.start(params)?;
|
||||
|
||||
// 用 self 的 clone（共享 Arc<Mutex> 状态）启动订阅，
|
||||
// 确保 register_auto_reconnect 注册的 listener 与 sub_handle 共享，
|
||||
// stop_subscription 时才能正确清理 listener。
|
||||
let kernel = self.clone();
|
||||
kernel.clone().register_auto_reconnect(app.clone()).await;
|
||||
let app_clone = app.clone();
|
||||
let handle = tauri::async_runtime::spawn(async move {
|
||||
kernel.start_subscription(app_clone).await;
|
||||
});
|
||||
*self.sub_handle.lock().await = Some(handle);
|
||||
|
||||
Ok(info)
|
||||
}
|
||||
}
|
||||
|
||||
/// 解析 SSE 事件文本，提取 data: 字段的 JSON 内容。
|
||||
/// 格式：`event: snapshot\ndata: {...json...}`
|
||||
fn parse_sse_data(event_str: &str) -> Option<String> {
|
||||
let mut data = String::new();
|
||||
for line in event_str.lines() {
|
||||
if let Some(rest) = line.strip_prefix("data:") {
|
||||
data.push_str(rest.trim());
|
||||
}
|
||||
}
|
||||
if data.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(data)
|
||||
}
|
||||
}
|
||||
|
||||
// ===================== Tauri 命令 =====================
|
||||
|
||||
#[tauri::command]
|
||||
pub fn monitor_kernel_info(
|
||||
state: tauri::State<'_, MonitorKernel>,
|
||||
app: AppHandle,
|
||||
) -> Result<MonitorKernelInfo, String> {
|
||||
state.prepare_kernel(&app)
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn monitor_status(
|
||||
state: tauri::State<'_, MonitorKernel>,
|
||||
pm: tauri::State<'_, ProcessManager>,
|
||||
) -> Result<MonitorStatus, String> {
|
||||
let info = pm.get_status(PROCESS_ID);
|
||||
let running = info
|
||||
.as_ref()
|
||||
.map(|i| matches!(i.status, crate::process_manager::ProcessStatus::Running))
|
||||
.unwrap_or(false);
|
||||
// 只在运行时查询 Kernel /status（避免未运行时发起无意义的 HTTP 请求）
|
||||
let kernel_status = if running { state.get_status().await.ok() } else { None };
|
||||
Ok(MonitorStatus {
|
||||
running,
|
||||
pid: info.as_ref().and_then(|i| i.pid),
|
||||
ready: kernel_status.as_ref().map(|s| s.ready).unwrap_or(false),
|
||||
sensor_count: kernel_status.as_ref().map(|s| s.sensor_count).unwrap_or(0),
|
||||
restart_count: info.as_ref().map(|i| i.restart_count).unwrap_or(0),
|
||||
})
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn monitor_start(
|
||||
state: tauri::State<'_, MonitorKernel>,
|
||||
app: AppHandle,
|
||||
) -> Result<crate::process_manager::ProcessInfo, String> {
|
||||
state.start_with_subscription(&app).await
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn monitor_stop(
|
||||
state: tauri::State<'_, MonitorKernel>,
|
||||
pm: tauri::State<'_, ProcessManager>,
|
||||
app: AppHandle,
|
||||
) -> Result<(), String> {
|
||||
state.stop_subscription(&app).await;
|
||||
pm.stop(PROCESS_ID)
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn monitor_get_status(state: tauri::State<'_, MonitorKernel>) -> Result<KernelStatus, String> {
|
||||
state.get_status().await
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn monitor_get_snapshot(state: tauri::State<'_, MonitorKernel>) -> Result<SensorSnapshot, String> {
|
||||
state.get_snapshot().await
|
||||
}
|
||||
Reference in New Issue
Block a user