RustWeightService/src/main.rs

use anyhow::{Context, Result};
use rumqttc::{Client, MqttOptions, QoS};
use serde::{Serialize, Deserialize};
use serde_json;
use std::io::{self, Read};
use std::time::Duration;
use std::thread;
use serialport;
use std::process::Command;
use actix_web::{web, App, HttpServer, HttpResponse};
use actix_cors::Cors;

#[derive(Serialize, Deserialize)]
struct WeightData {
    weight: String,
    timestamp: u64,
}

#[derive(Serialize)]
struct MacResponse {
    mac_address: String,
}

fn get_mac_address() -> String {
    let output = Command::new("getmac")
        .arg("/fo")
        .arg("csv")
        .arg("/nh")
        .output()
        .expect("Failed to execute getmac command");
    
    let output_str = String::from_utf8_lossy(&output.stdout);
    let first_line = output_str.lines().next().unwrap_or("");
    let mac = first_line.split(',').next().unwrap_or("").trim_matches('"');
    mac.replace('-', ":").to_lowercase()
}

// HTTP处理函数
async fn get_mac() -> HttpResponse {
    let mac = get_mac_address();
    println!("HTTP请求：获取MAC地址 = {}", mac);  // 添加日志
    let response = MacResponse {
        mac_address: mac,
    };
    HttpResponse::Ok().json(response)
}

#[actix_web::main]
async fn main() -> Result<()> {
    println!("程序启动...");
    
    // 启动HTTP服务器
    println!("正在启动HTTP服务器...");
    
    // 创建HTTP服务器
    let server = HttpServer::new(|| {
        // 配置CORS
        let cors = Cors::default()
            .allow_any_origin()  // 允许所有来源
            .allow_any_method()  // 允许所有HTTP方法
            .allow_any_header()  // 允许所有请求头
            .max_age(3600);      // 设置预检请求的缓存时间（秒）
            
        App::new()
            .wrap(cors)  // 添加CORS中间件
            .route("/mac", web::get().to(get_mac))
    })
    .bind(("0.0.0.0", 8080))?
    .run();
    
    println!("HTTP服务器已启动，监听在 http://127.0.0.1:8080");
    
    // 在新线程中运行MQTT和串口服务
    let mqtt_handle = tokio::spawn(async {
        if let Err(e) = run_mqtt_and_serial().await {
            eprintln!("MQTT/串口服务错误: {}", e);
        }
    });
    
    // 等待HTTP服务器结束
    server.await?;
    
    Ok(())
}

// MQTT和串口处理函数
async fn run_mqtt_and_serial() -> Result<()> {
    // 创建 MQTT 客户端
    let mut mqttopts = MqttOptions::new("weight_reader", "112.33.111.160", 1883);
    mqttopts.set_keep_alive(Duration::from_secs(5));
    mqttopts.set_credentials("auseft", "1q2w3E**");
    let (mut client, mut connection) = Client::new(mqttopts, 10);

    // 在单独的线程中处理 MQTT 连接
    thread::spawn(move || {
        for notification in connection.iter() {
            match notification {
                Ok(_) => {
                    // if(!notification.contains("PingRe")) {
                    //     println!("MQTT事件: {:?}", notification);
                    // }
                }
                Err(e) => {
                    eprintln!("MQTT连接错误: {:?}", e);
                    break;
                }
            }
        }
    });

    // List available ports
    let ports = serialport::available_ports().context("未找到串口设备!")?;
    
    if ports.is_empty() {
        println!("错误：没有找到任何串口设备");
        return Ok(());
    }

    println!("可用的串口设备:");
    for (i, port) in ports.iter().enumerate() {
        println!("{}: {}", i, port.port_name);
    }

    // Read port selection from user
    let port_idx = loop {
        println!("请输入要使用的串口编号 (0-{}):", ports.len() - 1);
        let mut input = String::new();
        io::stdin().read_line(&mut input)?;
        
        match input.trim().parse::<usize>() {
            Ok(idx) if idx < ports.len() => break idx,
            Ok(_) => println!("错误：输入的编号超出范围，请重新输入"),
            Err(_) => println!("错误：请输入有效的数字"),
        }
    };

    let port_name = &ports[port_idx].port_name;
    println!("正在尝试打开串口 {}...", port_name);

    // Configure and open the serial port
    let mut port = serialport::new(port_name, 9600)
        .timeout(Duration::from_millis(1000))
        .open()
        .with_context(|| format!("无法打开串口 {}", port_name))?;

    println!("成功打开串口 {}!", port_name);
    println!("正在读取数据... (按 Ctrl+C 退出)");

    // Read data continuously
    let mut serial_buf: Vec<u8> = vec![0; 1000];
    loop {
        match port.read(serial_buf.as_mut_slice()) {
            Ok(t) => {
                let data = String::from_utf8_lossy(&serial_buf[..t]);
                let trimmed_data = data.trim();
                if !trimmed_data.is_empty() {
                    println!("收到数据: {}", trimmed_data);
                    
                    // 创建权重数据结构
                    let weight_data = WeightData {
                        weight: trimmed_data.to_string(),
                        timestamp: std::time::SystemTime::now()
                            .duration_since(std::time::UNIX_EPOCH)
                            .unwrap()
                            .as_secs(),
                    };

                    // 开始循环发送数据
                    loop {
                        // 序列化数据
                        let json_data = serde_json::to_string(&weight_data)?;
                        println!("当前JSON数据: {}", json_data);
                        
                        // 获取MAC地址
                        let mac_address = get_mac_address();
                        println!("MAC地址: {}", mac_address);
                        // 发布到 MQTT，主题中包含MAC地址
                        if let Err(e) = client.publish(format!("weight/data/{}", mac_address), QoS::AtLeastOnce, false, json_data) {
                            eprintln!("MQTT发布错误: {:?}", e);
                        } else {
                            println!("成功发送数据到MQTT");
                        }

                        // 等待3秒
                        thread::sleep(Duration::from_secs(3));
                    }
                }
            }
            Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
                // Timeout is not an error, just continue
                continue;
            }
            Err(e) => {
                eprintln!("发生错误: {}", e);
                break;
            }
        }
    }

    Ok(())
}