302 lines
9.1 KiB
Rust
302 lines
9.1 KiB
Rust
|
//! 设备指令下发 — 通过 Redis 队列向设备发送指令
|
|||
|
|
//!
|
|||
|
|
//! API 服务不直接持有 TCP 连接，而是通过 Redis LIST `device:cmd:{imei}`
|
|||
|
|
//! 推送指令，由设备服务消费并发送到设备。指令响应通过 `device:reply:{imei}`
|
|||
|
|
//! 队列返回，API 服务通过 `PendingCommands` 匹配 msg_id 完成异步等待。
|
|||
|
|
|
|||
|
|
use std::collections::HashMap;
|
|||
|
|
use std::sync::Arc;
|
|||
|
|
use axum::extract::{Path, State};
|
|||
|
|
use axum::Json;
|
|||
|
|
use serde::Deserialize;
|
|||
|
|
use serde_json::{json, Value};
|
|||
|
|
use tokio::sync::{oneshot, Mutex};
|
|||
|
|
|
|||
|
|
use crate::error::AppError;
|
|||
|
|
use crate::middleware::auth::{self, CurrentUser};
|
|||
|
|
use crate::protocol::ServerResponse;
|
|||
|
|
|
|||
|
|
/// msg_id → oneshot Sender 映射表
|
|||
|
|
#[derive(Clone)]
|
|||
|
|
pub struct PendingCommands {
|
|||
|
|
inner: Arc<Mutex<HashMap<u32, oneshot::Sender<Value>>>>,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
impl PendingCommands {
|
|||
|
|
pub fn new() -> Self {
|
|||
|
|
Self { inner: Arc::new(Mutex::new(HashMap::new())) }
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 注册等待中的指令
|
|||
|
|
#[allow(dead_code)]
|
|||
|
|
pub async fn insert(&self, msg_id: u32, tx: oneshot::Sender<Value>) {
|
|||
|
|
self.inner.lock().await.insert(msg_id, tx);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 移除并返回指定 msg_id 的 sender（消费一次）
|
|||
|
|
pub async fn remove(&self, msg_id: u32) -> Option<oneshot::Sender<Value>> {
|
|||
|
|
self.inner.lock().await.remove(&msg_id)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 设备服务节点信息
|
|||
|
|
#[derive(Debug, Clone)]
|
|||
|
|
pub struct NodeInfo {
|
|||
|
|
pub node_id: String,
|
|||
|
|
pub ip: String,
|
|||
|
|
pub tcp_port: u16,
|
|||
|
|
/// RFC3339 格式的最后心跳时间，从 Redis 读取
|
|||
|
|
pub last_heartbeat: Option<String>,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 节点注册表（Redis 存储，API 服务从 Redis 读取节点信息）
|
|||
|
|
///
|
|||
|
|
/// 设备服务通过 HSET 写入 `node:{node_id}` hash，
|
|||
|
|
/// API 服务通过此模块读取。Redis 是节点信息的单一数据源。
|
|||
|
|
#[derive(Clone)]
|
|||
|
|
pub struct NodeRegistry {
|
|||
|
|
redis: redis::aio::ConnectionManager,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
impl NodeRegistry {
|
|||
|
|
pub fn new(redis: redis::aio::ConnectionManager) -> Self {
|
|||
|
|
Self { redis }
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
fn node_key(node_id: &str) -> String {
|
|||
|
|
format!("node:{}", node_id)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 注册节点（写入 Redis HSET）
|
|||
|
|
pub async fn register(&self, node_id: &str, ip: &str, tcp_port: u16) {
|
|||
|
|
let key = Self::node_key(node_id);
|
|||
|
|
let now = chrono::Utc::now().to_rfc3339();
|
|||
|
|
let mut conn = self.redis.clone();
|
|||
|
|
|
|||
|
|
let _: Result<(), _> = redis::cmd("HSET")
|
|||
|
|
.arg(&key)
|
|||
|
|
.arg("node_id").arg(node_id)
|
|||
|
|
.arg("ip").arg(ip)
|
|||
|
|
.arg("tcp_port").arg(tcp_port.to_string())
|
|||
|
|
.arg("status").arg("online")
|
|||
|
|
.arg("last_heartbeat").arg(&now)
|
|||
|
|
.query_async(&mut conn)
|
|||
|
|
.await;
|
|||
|
|
|
|||
|
|
// ponytail: HSET 不支持 EX，用 EXPIRE 设置 TTL
|
|||
|
|
let _: Result<(), _> = redis::cmd("EXPIRE")
|
|||
|
|
.arg(&key)
|
|||
|
|
.arg(300_u64)
|
|||
|
|
.query_async(&mut conn)
|
|||
|
|
.await;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 移除节点（从 Redis 删除）
|
|||
|
|
pub async fn remove(&self, node_id: &str) {
|
|||
|
|
let key = Self::node_key(node_id);
|
|||
|
|
let mut conn = self.redis.clone();
|
|||
|
|
let _: Result<(), _> = redis::cmd("DEL")
|
|||
|
|
.arg(&key)
|
|||
|
|
.query_async(&mut conn)
|
|||
|
|
.await;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 更新心跳时间戳
|
|||
|
|
pub async fn heartbeat(&self, node_id: &str) -> bool {
|
|||
|
|
let key = Self::node_key(node_id);
|
|||
|
|
let mut conn = self.redis.clone();
|
|||
|
|
let result: Result<(), _> = redis::cmd("HSET")
|
|||
|
|
.arg(&key)
|
|||
|
|
.arg("last_heartbeat")
|
|||
|
|
.arg(chrono::Utc::now().to_rfc3339())
|
|||
|
|
.query_async(&mut conn)
|
|||
|
|
.await;
|
|||
|
|
|
|||
|
|
// 心跳同时刷新 TTL
|
|||
|
|
let _: Result<(), _> = redis::cmd("EXPIRE")
|
|||
|
|
.arg(&key)
|
|||
|
|
.arg(300_u64)
|
|||
|
|
.query_async(&mut conn)
|
|||
|
|
.await;
|
|||
|
|
|
|||
|
|
result.is_ok()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 列出所有节点（从 Redis 读取，使用 SCAN 替代 KEYS 避免阻塞）
|
|||
|
|
pub async fn list(&self) -> Vec<NodeInfo> {
|
|||
|
|
let mut conn = self.redis.clone();
|
|||
|
|
let keys = self.scan_keys(&mut conn, "node:*").await;
|
|||
|
|
|
|||
|
|
let mut nodes = Vec::with_capacity(keys.len());
|
|||
|
|
for key in keys {
|
|||
|
|
let mut conn = self.redis.clone();
|
|||
|
|
let info: HashMap<String, String> = redis::cmd("HGETALL")
|
|||
|
|
.arg(&key)
|
|||
|
|
.query_async(&mut conn)
|
|||
|
|
.await
|
|||
|
|
.unwrap_or_default();
|
|||
|
|
|
|||
|
|
let node_id = info.get("node_id").cloned().unwrap_or_default();
|
|||
|
|
if node_id.is_empty() {
|
|||
|
|
continue; // ponytail: 跳过空记录
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
nodes.push(NodeInfo {
|
|||
|
|
node_id,
|
|||
|
|
ip: info.get("ip").cloned().unwrap_or_default(),
|
|||
|
|
tcp_port: info.get("tcp_port").and_then(|p| p.parse().ok()).unwrap_or(0),
|
|||
|
|
last_heartbeat: info.get("last_heartbeat").cloned(),
|
|||
|
|
});
|
|||
|
|
}
|
|||
|
|
nodes
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 使用 SCAN 命令安全地获取所有匹配模式的键（避免 KEYS 阻塞）
|
|||
|
|
async fn scan_keys(&self, conn: &mut redis::aio::ConnectionManager, pattern: &str) -> Vec<String> {
|
|||
|
|
let mut keys = Vec::new();
|
|||
|
|
let mut cursor = 0i64;
|
|||
|
|
loop {
|
|||
|
|
let (next_cursor, batch): (i64, Vec<String>) = redis::cmd("SCAN")
|
|||
|
|
.arg(cursor)
|
|||
|
|
.arg("MATCH")
|
|||
|
|
.arg(pattern)
|
|||
|
|
.arg("COUNT")
|
|||
|
|
.arg(100)
|
|||
|
|
.query_async(conn)
|
|||
|
|
.await
|
|||
|
|
.unwrap_or_default();
|
|||
|
|
keys.extend(batch);
|
|||
|
|
if next_cursor == 0 {
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
cursor = next_cursor;
|
|||
|
|
}
|
|||
|
|
keys
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 检查设备是否在线（通过 Redis 中的在线标记）
|
|||
|
|
pub async fn is_device_online(&self, imei: &str) -> bool {
|
|||
|
|
let key = format!("device:online:{}", imei);
|
|||
|
|
let mut conn = self.redis.clone();
|
|||
|
|
let exists: Result<bool, _> = redis::cmd("EXISTS")
|
|||
|
|
.arg(&key)
|
|||
|
|
.query_async(&mut conn)
|
|||
|
|
.await;
|
|||
|
|
exists.unwrap_or(false)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 应用共享状态
|
|||
|
|
#[derive(Clone)]
|
|||
|
|
#[allow(dead_code)]
|
|||
|
|
pub struct AppState {
|
|||
|
|
pub redis: redis::aio::ConnectionManager,
|
|||
|
|
pub mysql: sqlx::MySqlPool,
|
|||
|
|
pub pending_commands: PendingCommands,
|
|||
|
|
pub nodes: NodeRegistry,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 指令下发请求体
|
|||
|
|
#[derive(Debug, Deserialize)]
|
|||
|
|
pub struct CommandRequest {
|
|||
|
|
pub act: String,
|
|||
|
|
#[serde(default)]
|
|||
|
|
pub msg_id: u32,
|
|||
|
|
pub sub_device_id: Option<u8>,
|
|||
|
|
#[serde(flatten)]
|
|||
|
|
pub params: Option<Value>,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 设备指令结构（序列化后推入 Redis）
|
|||
|
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|||
|
|
pub struct DeviceCommand {
|
|||
|
|
pub act: String,
|
|||
|
|
pub msg_id: u32,
|
|||
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|||
|
|
pub dev_id: Option<String>,
|
|||
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|||
|
|
pub sub_device_id: Option<u8>,
|
|||
|
|
/// 扩展参数（开门指令、充电参数等）
|
|||
|
|
#[serde(flatten)]
|
|||
|
|
pub params: Option<Value>,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
impl DeviceCommand {
|
|||
|
|
/// 序列化为 JSON 字符串（紧凑格式）。
|
|||
|
|
pub fn to_json(&self) -> String {
|
|||
|
|
serde_json::to_string(self).expect("DeviceCommand 序列化不应失败")
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 向设备发送指令（通过 Redis 队列）
|
|||
|
|
pub async fn send_device_command(
|
|||
|
|
redis: &redis::aio::ConnectionManager,
|
|||
|
|
_dev_id: &str,
|
|||
|
|
cmd: &DeviceCommand,
|
|||
|
|
) -> Result<(), String> {
|
|||
|
|
let json_str = cmd.to_json();
|
|||
|
|
let mut conn = redis.clone();
|
|||
|
|
redis::cmd("LPUSH")
|
|||
|
|
.arg("device:commands")
|
|||
|
|
.arg(&json_str)
|
|||
|
|
.query_async::<()>(&mut conn)
|
|||
|
|
.await
|
|||
|
|
.map_err(|e| format!("Redis 推入失败: {}", e))?;
|
|||
|
|
Ok(())
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// 向设备发送响应（通过 Redis 队列）
|
|||
|
|
#[allow(dead_code)]
|
|||
|
|
pub async fn send_device_response(
|
|||
|
|
redis: &redis::aio::ConnectionManager,
|
|||
|
|
_dev_id: &str,
|
|||
|
|
resp: &ServerResponse,
|
|||
|
|
) -> Result<(), String> {
|
|||
|
|
let json_str = resp.to_json();
|
|||
|
|
let mut conn = redis.clone();
|
|||
|
|
redis::cmd("LPUSH")
|
|||
|
|
.arg("device:commands")
|
|||
|
|
.arg(&json_str)
|
|||
|
|
.query_async::<()>(&mut conn)
|
|||
|
|
.await
|
|||
|
|
.map_err(|e| format!("Redis 推入失败: {}", e))?;
|
|||
|
|
Ok(())
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// POST /api/devices/:dev_id/command — 向指定设备下发指令
|
|||
|
|
pub async fn send_command(
|
|||
|
|
user: CurrentUser,
|
|||
|
|
State(state): State<AppState>,
|
|||
|
|
Path(dev_id): Path<String>,
|
|||
|
|
Json(req): Json<CommandRequest>,
|
|||
|
|
) -> Result<Json<Value>, AppError> {
|
|||
|
|
auth::check_permission(&user, "device:operate")?;
|
|||
|
|
|
|||
|
|
let cmd = DeviceCommand {
|
|||
|
|
act: req.act,
|
|||
|
|
msg_id: req.msg_id,
|
|||
|
|
dev_id: Some(dev_id.clone()),
|
|||
|
|
sub_device_id: req.sub_device_id,
|
|||
|
|
params: req.params,
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
send_device_command(&state.redis, &dev_id, &cmd)
|
|||
|
|
.await
|
|||
|
|
.map_err(AppError::Internal)?;
|
|||
|
|
|
|||
|
|
tracing::info!("[command] 已推入队列 dev_id={} act={}", dev_id, cmd.act);
|
|||
|
|
Ok(Json(json!({ "suc": 1 })))
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/// GET /api/devices/stats — 在线设备统计
|
|||
|
|
pub async fn device_stats(
|
|||
|
|
user: CurrentUser,
|
|||
|
|
State(state): State<AppState>,
|
|||
|
|
) -> Result<Json<Value>, AppError> {
|
|||
|
|
auth::check_permission(&user, "device:view")?;
|
|||
|
|
let nodes = state.nodes.list().await;
|
|||
|
|
Ok(Json(json!({
|
|||
|
|
"node_count": nodes.len(),
|
|||
|
|
})))
|
|||
|
|
}
|