//! 设备消息处理逻辑(轻量化)。 //! //! 设备服务只做**协议透传**,不处理业务逻辑: //! - `status_post` 的通道数据在设备侧已是 **18 字符 hex 明文**,本服务**原样透传**, //! 不再做任何拆解/位域解析(解析归 api-server 的业务层) //! - 所有设备上报消息统一转发到 Redis `device:reports` 队列 //! - 不写 DB、不验证签名、不处理业务规则 //! - auth_str/login 等协议级消息也由 API 服务处理后通过 `device:commands` 回复 use super::connection::ConnectionPool; use super::protocol::DeviceMessage; /// 分发消息:补写设备身份后转发到 Redis 上报队列。 /// /// `current_dev_id` = 本 TCP 连接绑定的设备身份(由 `auth_str`/`login` 建立)。 /// 为什么需要它:新协议里业务上行**不带 `dev_id`**(设备身份由连接确定), /// 而 api-server 依赖 `dev_id` 识别设备,故转发前必须补上。 /// /// 本服务**不直接回复设备**:应答由 api-server 经 Redis `device:commands` 下发, /// 因此这里没有返回值(历史上曾返回 `Option` 但恒为 `None`, /// 那个结构体副本已删除,避免与 api-server 的真实定义漂移)。 pub async fn handle_message( mut msg: DeviceMessage, current_dev_id: Option<&str>, _pool: &ConnectionPool, redis: &redis::aio::ConnectionManager, ) { attach_dev_id(&mut msg, current_dev_id); // 所有消息统一转发到 Redis,由 API 服务的 report_worker 处理 forward_to_redis(redis, &msg).await; // 不直接回复设备,等待 API 服务通过 device:commands 队列回复 } /// 把连接绑定的设备身份写入待转发报文。 /// /// 以**连接**为准(而非报文自带值):设备身份由 TCP 连接确定, /// 这样即使设备漏带/错带 `dev_id`,api-server 侧也不会认错设备。 fn attach_dev_id(msg: &mut DeviceMessage, current_dev_id: Option<&str>) { match current_dev_id { Some(dev_id) if !dev_id.is_empty() => msg.dev_id = Some(dev_id.to_string()), // 理论上不会出现:未登录连接的业务报文已被 server.rs 的门控丢弃。 // 此时仍照常转发(不丢数据),仅记 WARN 便于排查。 _ => tracing::warn!( "[forward] 连接未绑定 dev_id,转发报文将不含设备身份, act={}", msg.act ), } } /// 将设备消息转发到 Redis 统一上报队列 `device:reports` async fn forward_to_redis(redis: &redis::aio::ConnectionManager, msg: &DeviceMessage) { let json_str = match serde_json::to_string(msg) { Ok(json) => json, Err(e) => { // 序列化失败说明报文结构异常,记 ERROR 并丢弃(不推送半截数据) tracing::error!("[forward] 消息序列化失败 act={} err={}", msg.act, e); return; } }; let mut conn = redis.clone(); let result: Result<(), _> = redis::cmd("LPUSH") .arg("device:reports") .arg(&json_str) .query_async(&mut conn) .await; match result { Ok(_) => { tracing::debug!( "[forward] 已推送 dev_id={} act={}", msg.dev_id.as_deref().unwrap_or("?"), msg.act ); } Err(e) => tracing::error!("[forward] Redis LPUSH 失败: {}", e), } }