Acasă Explorează Ajutor
Înregistrare Autentificare
lee
/
zGrid-api
1
0
Bifurcare 0
Fisiere Probleme 0 Trageți solicitările 0 Wiki
Răsfoiți Sursa

Get MQTT up and working

Lee Morgan 2 luni în urmă
părinte
c58b9a5157
comite
81f23704e4
4 a modificat fișierele cu 44 adăugiri și 3 ștergeri
Vizualizare divizată Arată Statisticile Diff
  1. 3 2
      Cargo.lock
  2. 1 0
      Cargo.toml
  3. 19 1
      src/controllers/device/signal.rs
  4. 21 0
      src/main.rs

+ 3 - 2
Cargo.lock
Vizualizați fișierul 

@@ -1032,6 +1032,7 @@ dependencies = [
 
  "serde_json",
 
  "sqlx",
 
  "thiserror",
 
+ "tokio",
 
  "uuid",
 
 ]
 
 
@@ -2474,9 +2475,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 
 [[package]]
 
 name = "tokio"
 
-version = "1.52.1"
 
+version = "1.52.3"
 
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
-checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6"
 
+checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe"
 
 dependencies = [
 
  "bytes",
 
  "libc",

+ 1 - 0
Cargo.toml
Vizualizați fișierul 

@@ -20,4 +20,5 @@ sqlx = { version = "0.8.6", features = [
 
     "uuid"
 
 ]}
 
 thiserror = "2.0.18"
 
+tokio = "1.52.3"
 
 uuid = { version="1.23.1", features=["v4", "v7", "serde"] }

+ 19 - 1
src/controllers/device/signal.rs
Vizualizați fișierul 

@@ -3,6 +3,7 @@ use serde::Deserialize;
 
 use sqlx::PgPool;
 
 use uuid::Uuid;
 
 use serde_json::json;
 
+use rumqttc::{AsyncClient, QoS};
 
 use crate::{
 
     auth::user_auth,
 
     app_error::AppError,
 
@@ -23,6 +24,7 @@ struct Params {
 
 #[post("/home/{home_id}/device/{device_id}/signal")]
 
 pub async fn route(
 
     db: web::Data<PgPool>,
 
+    mqtt: web::Data<AsyncClient>,
 
     body: web::Json<Body>,
 
     params: web::Path<Params>,
 
     req: HttpRequest
 
@@ -32,14 +34,30 @@ pub async fn route(
 
     let device_id = Uuid::parse_str(&params.device_id)?;
 
     let device = Device::find_one(&db, user.id, home_id, device_id).await?;
 
     validate_signal(&device.device_type, &body.signal)?;
 
+    send_signal(device, &body.signal, &mqtt).await?;
 
     Ok(HttpResponse::Ok().json(json!({"success": true})))
 
 }
 
 
 fn validate_signal(device_type: &DeviceType, signal: &String) -> Result<(), AppError> {
 
     match device_type {
 
         DeviceType::OnOffSwitch => match signal.as_str() {
 
-            "on" | "off" => Ok(()),
 
+            "ON" | "OFF" => Ok(()),
 
             _ => Err(AppError::invalid_input("Invalid signal for 'on_off_switch'"))
 
         }
 
     }
 
 }
 
+
 
+async fn send_signal(device: Device, signal: &String, client: &AsyncClient) -> Result<(), AppError> {
 
+    let topic = format!("zigbee2mqtt/{}/set", device.name);
 
+    let payload = json!({"state": signal});
 
+
 
+    match client.publish(
 
+        &topic,
 
+        QoS::AtLeastOnce,
 
+        false,
 
+        payload.to_string().as_bytes()
 
+    ).await {
 
+        Ok(_) => Ok(()),
 
+        Err(e) => Err(AppError::InternalError(e.to_string()))
 
+    }
 
+}

+ 21 - 0
src/main.rs
Vizualizați fișierul 

@@ -1,6 +1,7 @@
 
 use actix_web::{HttpServer, web, middleware, App};
 
 use actix_cors::Cors;
 
 use sqlx::PgPool;
 
+use rumqttc::{MqttOptions, AsyncClient};
 
 use crate::app_error::AppError;
 
 
 mod app_error;
 
@@ -12,6 +13,25 @@ mod auth;
 
 
 #[actix_web::main]
 
 async fn main() -> Result<(), AppError> {
 
+    //MQTT
 
+    let mut options = MqttOptions::new("actix_server", "localhost", 1883);
 
+    options.set_keep_alive(std::time::Duration::from_secs(10));
 
+    let (client, mut connection) = AsyncClient::new(options, 10);
 
+
 
+    tokio::spawn(async move {
 
+        loop {
 
+            match connection.poll().await {
 
+                Ok(_) => {},
 
+                Err(e) => {
 
+                    eprintln!("MQTT error: {}", e);
 
+                }
 
+            }
 
+        }
 
+    });
 
+
 
+    let mqtt_client = client;
 
+
 
+    //Postgres
 
     let app_env = std::env::var("APP_ENV").unwrap_or_else(|_| "development".to_string());
 
     let pool = PgPool::connect("postgres://leemorgan:password123@localhost:5432/homo")
 
         .await
 
@@ -34,6 +54,7 @@ async fn main() -> Result<(), AppError> {
 
             .wrap(middleware::Compress::default())
 
             .wrap(cors)
 
             .app_data(web::Data::new(pool.clone()))
 
+            .app_data(web::Data::new(mqtt_client.clone()))
 
             .app_data(
 
                 web::JsonConfig::default().error_handler(|err, _req| {
 
                     AppError::JsonDeserializationError(err.to_string()).into()