Merge pull request #18 from vxshug/main

vxshug · web-flow · commit e2fac314ab51 · 2025-06-30T10:41:21.000+08:00
fix active time
diff --git a/devices_manager/src/man/mqtt.rs b/devices_manager/src/man/mqtt.rs
@@ -1,26 +1,25 @@
-use std::str::FromStr;
-use std::string::FromUtf8Error;
+use crate::event::config_cache;
 use crate::load::{load_config, MqttConfig};
+use crate::man::data::DownloadData;
+use crate::man::gw::{DataWrapper, GwCmd, GwCmdResponse, ShellCmd};
+use crate::man::lora::{LoRaNode, LoRaNodeManager};
+use crate::man::Id;
+use crate::protocol::mqtt::{LinkRx, LinkTx, Notification};
 use crate::{DeviceError, DeviceResult, GLOBAL_DOWNLOAD_RESPONSE, GLOBAL_STATE};
-use derive_new::new;
-use rumqttc::{Event, Incoming, MqttOptions, QoS};
-use std::sync::Mutex;
-use std::time::Duration;
 use base64::Engine;
 use bytes::Bytes;
+use common_define::db::{DbErr, DeviceLoraGateColumn, DeviceLoraGateEntity, Eui};
+use common_define::event::DownloadMessage;
+use derive_new::new;
+use rumqttc::{Event, Incoming, MqttOptions, QoS};
 use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
 use serde::{Deserialize, Serialize};
+use std::str::FromStr;
+use std::string::FromUtf8Error;
+use std::sync::Mutex;
+use std::time::Duration;
 use tokio::sync::mpsc;
 use tracing::{debug, error, info, trace, warn};
-use common_define::db::{DbErr, DeviceLoraGateColumn, DeviceLoraGateEntity, Eui};
-use common_define::event::DownloadMessage;
-use crate::event::config_cache;
-use crate::man::data::DownloadData;
-use crate::man::gw::{DataWrapper, GwCmd, GwCmdResponse, ShellCmd};
-use crate::man::Id;
-use crate::man::lora::{LoRaNode, LoRaNodeManager};
-use crate::protocol::mqtt::{LinkRx, LinkTx, Notification};
-
 
 #[derive(Debug, Deserialize, Serialize)]
 pub struct ForwardResult {
@@ -74,7 +73,6 @@ pub enum MqttError {
 }
 
 impl MqPublisher {
-
     pub fn new(client: LinkTx) -> MqPublisher {
         MqPublisher { client: Mutex::new(client) }
     }
@@ -88,21 +86,32 @@ impl MqPublisher {
 pub struct MessageProcessor {
     rx: LinkRx,
     sender: mpsc::Sender<MqttMessage>,
-    down: mpsc::Sender<DownloadMessage>
+    down: mpsc::Sender<DownloadMessage>,
 }
 
 impl MessageProcessor {
-    pub fn new_with_sender(rx: LinkRx, sender: mpsc::Sender<MqttMessage>, down: mpsc::Sender<DownloadMessage>) -> Self {
+    pub fn new_with_sender(
+        rx: LinkRx,
+        sender: mpsc::Sender<MqttMessage>,
+        down: mpsc::Sender<DownloadMessage>,
+    ) -> Self {
         Self { sender, rx, down }
     }
 
-
     pub async fn start(mut self) {
         while let Ok(message) = self.rx.next().await {
             if let Some(Notification::Forward(publish)) = message {
+                let down = self.down.clone();
                 match String::from_utf8(publish.publish.topic.to_vec()) {
                     Ok(topic) => {
-                        tokio::spawn(process_mqtt(MqttMessage::new(topic, publish.publish.payload), self.down.clone()));
+                        tokio::spawn(async move {
+                            if let Err(e) =
+                                process_mqtt(MqttMessage::new(topic, publish.publish.payload), down)
+                                    .await
+                            {
+                                warn!("mqtt payload: {}", e)
+                            }
+                        });
                     }
                     Err(_) => {
                         warn!("Snap Mqtt message contained invalid UTF-8");
@@ -113,7 +122,10 @@ impl MessageProcessor {
     }
 }
 
-async fn process_mqtt(message: MqttMessage, down: mpsc::Sender<DownloadMessage>) -> Result<(), MqttError> {
+async fn process_mqtt(
+    message: MqttMessage,
+    down: mpsc::Sender<DownloadMessage>,
+) -> Result<(), MqttError> {
     let mut topic = message.topic.splitn(3, '/');
     topic.next();
     if let Some(user_id) = topic.next() {
@@ -130,8 +142,12 @@ async fn process_mqtt(message: MqttMessage, down: mpsc::Sender<DownloadMessage>)
     }
     Ok(())
 }
-async fn process_downlink(user_id: Id, message: MqttMessage, down: mpsc::Sender<DownloadMessage>) -> Result<(), MqttError> {
-    let mut topic = message.topic.split( '/');
+async fn process_downlink(
+    user_id: Id,
+    message: MqttMessage,
+    down: mpsc::Sender<DownloadMessage>,
+) -> Result<(), MqttError> {
+    let mut topic = message.topic.split('/');
     topic.next();
     topic.next();
     topic.next();
@@ -142,15 +158,11 @@ async fn process_downlink(user_id: Id, message: MqttMessage, down: mpsc::Sender<
         if let Some(node) = LoRaNodeManager::get_node_by_eui(eui).await? {
             if node.info.user_id == Some(user_id) {
                 let data: ForwardPayload = serde_json::from_slice(&message.payload)?;
-                down.send(DownloadMessage {
-                    eui,
-                    port: data.port,
-                    data: data.payload,
-                }).await;
+                down.send(DownloadMessage { eui, port: data.port, data: data.payload }).await;
                 let rx = GLOBAL_DOWNLOAD_RESPONSE.add(eui);
                 let topic = format!("user/{}/device/{}/forward_result", user_id, eui);
                 if data.timeout > 60 {
-                    return Ok(())
+                    return Ok(());
                 }
                 match rx {
                     Some(rx) => {
@@ -159,10 +171,16 @@ async fn process_downlink(user_id: Id, message: MqttMessage, down: mpsc::Sender<
                             Ok(Ok(response)) => {
                                 let payload = serde_json::to_string(&ForwardResult {
                                     port: response.1,
-                                    payload: base64::engine::general_purpose::STANDARD.encode(response.0.as_slice()),
-                                    status: 0
+                                    payload: base64::engine::general_purpose::STANDARD
+                                        .encode(response.0.as_slice()),
+                                    status: 0,
                                 })?;
-                                GLOBAL_STATE.mq.publish(crate::integration::mqtt::MqttMessage::new(payload, topic)).await?;
+                                GLOBAL_STATE
+                                    .mq
+                                    .publish(crate::integration::mqtt::MqttMessage::new(
+                                        payload, topic,
+                                    ))
+                                    .await?;
                             }
                             _ => {
                                 GLOBAL_DOWNLOAD_RESPONSE.get(eui);
@@ -173,9 +191,12 @@ async fn process_downlink(user_id: Id, message: MqttMessage, down: mpsc::Sender<
                         let payload = serde_json::to_string(&ForwardResult {
                             port: 0,
                             payload: "".to_string(),
-                            status: -1
+                            status: -1,
                         })?;
-                        GLOBAL_STATE.mq.publish(crate::integration::mqtt::MqttMessage::new(payload, topic)).await?;
+                        GLOBAL_STATE
+                            .mq
+                            .publish(crate::integration::mqtt::MqttMessage::new(payload, topic))
+                            .await?;
                     }
                 }
             }
@@ -184,16 +205,20 @@ async fn process_downlink(user_id: Id, message: MqttMessage, down: mpsc::Sender<
     Ok(())
 }
 async fn process_gateway(user_id: Id, message: MqttMessage) -> Result<(), MqttError> {
-    let mut topic = message.topic.split( '/');
+    let mut topic = message.topic.split('/');
     topic.next();
     topic.next();
     topic.next();
     let eui_s = topic.next().ok_or(MqttError::EuiNotFound)?;
     let action = topic.next().ok_or(MqttError::ActionNotFound)?;
     if "up" == action {
-        let cmd: GwCmdResponse = serde_json::from_slice(message.payload.as_ref()).map_err(|e| MqttError::SerdeError(e))?;
+        let cmd: GwCmdResponse = serde_json::from_slice(message.payload.as_ref())
+            .map_err(|e| MqttError::SerdeError(e))?;
         let eui = Eui::from_str(eui_s)?;
-        let gate = DeviceLoraGateEntity::find().filter(DeviceLoraGateColumn::Eui.eq(eui)).one(&GLOBAL_STATE.db).await?;
+        let gate = DeviceLoraGateEntity::find()
+            .filter(DeviceLoraGateColumn::Eui.eq(eui))
+            .one(&GLOBAL_STATE.db)
+            .await?;
         if let Some(gate) = gate {
             match cmd {
                 GwCmdResponse::ShellCmd(DataWrapper { id, data }) => {}
diff --git a/devices_manager/src/protocol/mqtt/server/router.rs b/devices_manager/src/protocol/mqtt/server/router.rs
@@ -435,7 +435,7 @@ impl Router {
         assert_eq!(self.obufs.insert(outgoing), connection_id);
 
         self.connection_map.insert(client_id.clone(), connection_id);
-        info!(connection_id, "Client connection registered");
+        debug!(connection_id, "Client connection registered");
 
         assert_eq!(self.ackslog.insert(ackslog), connection_id);
         assert_eq!(self.scheduler.add(tracker), connection_id);
@@ -731,7 +731,7 @@ impl Router {
                             tracing::info_span!("subscribe", topic = f.path, pkid = subscribe.pkid);
                         let _guard = span.enter();
 
-                        info!("Adding subscription on topic {}", f.path);
+                        debug!("Adding subscription on topic {}", f.path);
                         let connection = self.connections.get_mut(id).unwrap();
 
                         if let Err(e) = validate_subscription(connection, f) {
diff --git a/devices_manager/src/protocol/mqtt/server/server.rs b/devices_manager/src/protocol/mqtt/server/server.rs
@@ -9,6 +9,7 @@ use tokio::net::{TcpListener, TcpStream};
 use tokio::time;
 use tokio::time::error::Elapsed;
 use tracing::{error, info, field, Instrument, Span};
+use tracing::log::debug;
 use uuid::Uuid;
 use ws_stream_tungstenite::WsStream;
 use crate::protocol::mqtt::{version, ConnectionId};
@@ -127,7 +128,7 @@ impl Server {
                 };
             }
 
-            info!(
+            debug!(
                 name=?self.config.name, ?addr, count, tenant=?tenant_id, "accept"
             );
 
diff --git a/migration/src/m20240904_020441_create_table.rs b/migration/src/m20240904_020441_create_table.rs
@@ -227,6 +227,7 @@ mod device {
                     .col(big_key_auto(SnapIntegrationMqtt::Id))
                     .col(big_integer(SnapIntegrationMqtt::UserId))
                     .col(text(SnapIntegrationMqtt::MqttType))
+                    .col(text(SnapIntegrationMqtt::Name))
                     .col(text(SnapIntegrationMqtt::Username))
                     .col(text(SnapIntegrationMqtt::Password))
                     .col(boolean(SnapIntegrationMqtt::Enable))
@@ -500,8 +501,9 @@ mod device {
     enum SnapIntegrationMqtt {
         Table,
         Id,
-        MqttType,
         UserId,
+        MqttType,
+        Name,
         Username,
         Password,
         Enable,
diff --git a/snap_api/src/service/device/device.rs b/snap_api/src/service/device/device.rs
@@ -1,4 +1,12 @@
-use common_define::db::{DecodeScriptColumn, DecodeScriptEntity, DeviceAuthorityActiveModel, DeviceAuthorityColumn, DeviceAuthorityEntity, DeviceAuthorityModel, DeviceDataEntity, DeviceDataModel, DeviceFunctionColumn, DeviceFunctionEntity, DeviceFunctionModel, DeviceLoraGateActiveModel, DeviceLoraGateColumn, DeviceLoraGateEntity, DeviceLoraGateModel, DeviceLoraNodeColumn, DeviceLoraNodeEntity, DeviceLoraNodeModel, DevicesActiveModel, DevicesColumn, DevicesEntity, DevicesModel, Eui, Key, LoRaAddr, SnapDeviceColumn, SnapDeviceDataNameColumn, SnapDeviceDataNameEntity, SnapDeviceEntity, SnapDeviceModel};
+use common_define::db::{
+    DecodeScriptColumn, DecodeScriptEntity, DeviceAuthorityActiveModel, DeviceAuthorityColumn,
+    DeviceAuthorityEntity, DeviceAuthorityModel, DeviceDataEntity, DeviceDataModel,
+    DeviceFunctionColumn, DeviceFunctionEntity, DeviceFunctionModel, DeviceLoraGateActiveModel,
+    DeviceLoraGateColumn, DeviceLoraGateEntity, DeviceLoraGateModel, DeviceLoraNodeColumn,
+    DeviceLoraNodeEntity, DeviceLoraNodeModel, DevicesActiveModel, DevicesColumn, DevicesEntity,
+    DevicesModel, Eui, Key, LoRaAddr, SnapDeviceColumn, SnapDeviceDataNameColumn,
+    SnapDeviceDataNameEntity, SnapDeviceEntity, SnapDeviceModel,
+};
 use common_define::decode::LastDecodeData;
 use common_define::lora::{LoRaJoinType, LoRaRegion};
 use common_define::product::{DeviceType, ProductType, ShareType};
@@ -16,20 +24,20 @@ use serde_json::Value;
 use std::collections::HashMap;
 use tracing::{debug, instrument};
 
+use super::DeviceService;
 use crate::cache::DeviceCache;
 use crate::error::ApiError;
 use crate::error::ApiResult;
-use crate::{
-    get_lang, tt, AppState, CurrentUser, DEVICE_DATA_RAW_SQL, MODEL_MAP, SEA_ORMDB_BACKEND,
-};
 use crate::gw_conf::GwConfig;
-use super::DeviceService;
 use crate::service::data::query::{DataDeviceOneResponse, TimeDate};
 use crate::service::data::DataService;
 use crate::service::device::group::{DeviceGroupResp, DeviceGroupService};
 use crate::service::lorawan::{LoRaGateService, LoRaNodeService, ReqLoraGateway, ReqLoraNode};
 use crate::service::mqtt::{MQTTService, ReqMQTT};
 use crate::service::snap::{ReqSnap, SnapDeviceService, SnapJoinParameter};
+use crate::{
+    get_lang, tt, AppState, CurrentUser, DEVICE_DATA_RAW_SQL, MODEL_MAP, SEA_ORMDB_BACKEND,
+};
 
 #[derive(Serialize)]
 pub(crate) struct DeviceIoResp {
@@ -628,31 +636,24 @@ impl DeviceService {
         redis: &mut R,
         conn: &C,
     ) -> ApiResult<DeviceCache> {
-        match DeviceCache::load_by_user_id(user_id, redis).await? {
-            Some(cache) => Ok(cache),
-            None => {
-                let devices = DeviceAuthorityEntity::find()
-                    .filter(
-                        DeviceAuthorityColumn::ShareId
-                            .eq(user_id)
-                            .and(DeviceAuthorityColumn::ShareType.eq(ShareType::User.as_ref())),
-                    )
-                    .find_also_related(DevicesEntity)
-                    .all(conn)
-                    .await?;
-                let devices = devices
-                    .into_iter()
-                    .filter_map(|item| match item.1 {
-                        Some(device) => Some(DeviceWithAuth { device, auth: item.0 }),
-                        None => None,
-                    })
-                    .collect::<Vec<_>>();
-                let device_cache = DeviceCache::new(devices);
-                debug!("update device cache");
-                device_cache.save_by_user_id(user_id, redis).await?;
-                Ok(device_cache)
-            }
-        }
+        let devices = DeviceAuthorityEntity::find()
+            .filter(
+                DeviceAuthorityColumn::ShareId
+                    .eq(user_id)
+                    .and(DeviceAuthorityColumn::ShareType.eq(ShareType::User.as_ref())),
+            )
+            .find_also_related(DevicesEntity)
+            .all(conn)
+            .await?;
+        let devices = devices
+            .into_iter()
+            .filter_map(|item| match item.1 {
+                Some(device) => Some(DeviceWithAuth { device, auth: item.0 }),
+                None => None,
+            })
+            .collect::<Vec<_>>();
+        let device_cache = DeviceCache::new(devices);
+        Ok(device_cache)
     }
     pub(crate) async fn query_all_with_ids<C: ConnectionTrait>(
         user_id: Id,
@@ -922,7 +923,6 @@ impl DeviceService {
             let mut gate = gate.into_active_model();
             if let Some(region) = info.region {
                 gate.region = ActiveValue::Set(region);
-
             }
             gate.update(conn).await?;
         }
diff --git a/snap_api/src/service/user/auth.rs b/snap_api/src/service/user/auth.rs
@@ -133,7 +133,7 @@ pub(crate) async fn auth(
                     }
                 }
             }
-            info!("auth success: {}", user.id);
+            debug!("request user id: {}, username: {}, path: {}", user.id, user.name, req.uri().path_and_query().map(|x| x.as_str()).unwrap_or_default());
             Ok(run_with_user(user.id, user.name, next.run(req)).await)
         }
         Err(e) => Ok(e.into_response()),
diff --git a/snap_proto/build.rs b/snap_proto/build.rs
@@ -17,6 +17,7 @@ fn main() {
         .field_attribute("manager.GwConfig.trust_mode", r#"#[serde(skip_serializing_if = "Option::is_none")]"#)
         .field_attribute("manager.GwConfig.api_token", r#"#[serde(skip_serializing_if = "Option::is_none")]"#)
         .field_attribute("manager.GwConfig.cert", r#"#[serde(skip_serializing_if = "Option::is_none")]"#)
+        .field_attribute("manager.GwConfig.client_cert", r#"#[serde(skip_serializing_if = "Option::is_none")]"#)
         .compile_protos(&["proto/manager.proto"], &["proto"])
         .unwrap();
 }

Original file line number	Diff line number	Diff line change
`@@ -133,7 +133,7 @@ pub(crate) async fn auth(`
`133`	`133`	`}`
`134`	`134`	`}`
`135`	`135`	`}`
`136`		`- info!("auth success: {}", user.id);`
	`136`	`+ debug!("request user id: {}, username: {}, path: {}", user.id, user.name, req.uri().path_and_query().map(\|x\| x.as_str()).unwrap_or_default());`
`137`	`137`	`Ok(run_with_user(user.id, user.name, next.run(req)).await)`
`138`	`138`	`}`
`139`	`139`	`Err(e) => Ok(e.into_response()),`
Original file line number	Diff line number	Diff line change
`@@ -17,6 +17,7 @@ fn main() {`
`17`	`17`	`.field_attribute("manager.GwConfig.trust_mode", r#"#[serde(skip_serializing_if = "Option::is_none")]"#)`
`18`	`18`	`.field_attribute("manager.GwConfig.api_token", r#"#[serde(skip_serializing_if = "Option::is_none")]"#)`
`19`	`19`	`.field_attribute("manager.GwConfig.cert", r#"#[serde(skip_serializing_if = "Option::is_none")]"#)`
	`20`	`+ .field_attribute("manager.GwConfig.client_cert", r#"#[serde(skip_serializing_if = "Option::is_none")]"#)`
`20`	`21`	`.compile_protos(&["proto/manager.proto"], &["proto"])`
`21`	`22`	`.unwrap();`
`22`	`23`	`}`