48 lines
1.7 KiB
Rust
48 lines
1.7 KiB
Rust
|
//! 指令响应处理协程
|
||
|
|
//!
|
||
|
|
//! 从 Redis LIST `device:reply:{imei}` 读取设备指令响应，
|
||
|
|
//! 匹配 msg_id 并通过 oneshot channel 通知等待中的 HTTP 请求。
|
||
|
|
|
||
|
|
use redis::aio::ConnectionManager;
|
||
|
|
|
||
|
|
use crate::commands::PendingCommands;
|
||
|
|
|
||
|
|
/// 启动指令响应处理协程
|
||
|
|
///
|
||
|
|
/// 使用 BRPOP 阻塞等待所有 `device:reply:*` 队列的响应消息。
|
||
|
|
pub async fn start(redis: ConnectionManager, pending: PendingCommands) {
|
||
|
|
tracing::info!("[reply_worker] 启动");
|
||
|
|
|
||
|
|
loop {
|
||
|
|
let mut conn = redis.clone();
|
||
|
|
let result: Result<(String, String), redis::RedisError> = redis::cmd("BRPOP")
|
||
|
|
.arg("device:replies")
|
||
|
|
.arg(0)
|
||
|
|
.query_async(&mut conn)
|
||
|
|
.await;
|
||
|
|
|
||
|
|
match result {
|
||
|
|
Ok((_key, reply_json)) => {
|
||
|
|
// 解析 msg_id
|
||
|
|
if let Ok(reply) = serde_json::from_str::<serde_json::Value>(&reply_json) {
|
||
|
|
let msg_id = reply["msg_id"].as_u64().unwrap_or(0) as u32;
|
||
|
|
let pending = pending.clone();
|
||
|
|
|
||
|
|
tokio::spawn(async move {
|
||
|
|
if let Some(tx) = pending.remove(msg_id).await {
|
||
|
|
let _ = tx.send(reply);
|
||
|
|
tracing::debug!("[reply_worker] msg_id={} 已匹配", msg_id);
|
||
|
|
} else {
|
||
|
|
tracing::debug!("[reply_worker] msg_id={} 无等待者（可能已超时）", msg_id);
|
||
|
|
}
|
||
|
|
});
|
||
|
|
}
|
||
|
|
}
|
||
|
|
Err(e) => {
|
||
|
|
tracing::error!("[reply_worker] BRPOP 错误: {}", e);
|
||
|
|
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|