车联网实战：TDMQ MQTT + 云函数打造车辆状态智能响应链路

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let vin = "VIN001";
    // 1. 拼接 MQTT 接入点（mqtts 表示走 TLS）
    let access_point = format!(
        "mqtts://mqtt-xxx.tencenttdmq.com:8883?client_id={}", vin
    );
    let mut opts = MqttOptions::parse_url(&access_point)?;
    opts.set_keep_alive(Duration::from_secs(30));   // 30s 心跳保活
    opts.set_credentials("root", "password");        // 鉴权（生产环境请用密钥）
    // 2. 启动模拟器：内部循环上报 + 订阅控制指令
    let mut emu = TcuEmulator::new(vin.to_owned(), opts);
    emu.run().await?;
    Ok(())
}

pub async fn run(&mut self) -> anyhow::Result<()> {
    // 订阅控制指令 Topic（接收云端下发的 ac_power 指令）
    let command_topic = format!("sh/tcu/{}/command", self.vin);
    self.client.subscribe(command_topic, QoS::AtLeastOnce).await?;
    while self.battery > 0 {
        tokio::time::sleep(Duration::from_secs(1)).await;
        // 电量按当前空调档位递减，模拟真实耗电
        self.battery -= self.ac_power.load(Ordering::Relaxed);
        let status = BatteryStatus { battery: self.battery };
        // 发布到 sh/tcu/VIN001/battery（此处省略发布细节）
    }
    Ok(())
}

fn process_event(event: Event, ac_power: &Arc<AtomicU8>) {
    if let Incoming(Packet::Publish(publish)) = event {
        if let Ok(cmd) = serde_json::from_slice::<Command>(&publish.payload) {
            info!("收到控制指令: {:?}", cmd);
            // 用原子变量更新空调档位，主循环下一秒生效
            ac_power.store(cmd.ac_power, Ordering::Relaxed);
        }
    }
}

public class Main {
    public String mainHandler(MqttEvent event, Context context) {
        logger.info("request-id={}", context.getRequestId());
        trySavePower(event);
        return objectMapper.writeValueAsString(event);
    }
}

private static void trySavePower(MqttEvent event) {
    for (Record record : event.getRecords()) {
        BatteryStatus status = objectMapper.readValue(
            record.getBody(), BatteryStatus.class);
        if (status.getBattery() <= 10) {
            turnOffAc(record);       // ≤ 10%：关闭空调，保续航
        } else if (status.getBattery() <= 30) {
            savePower(record);       // ≤ 30%：降低空调功率
        }
    }
}

static void publish(Record record, int targetAcPower) {
    String host = System.getenv("HOST");
    // 由上行 Topic 推导下行 Topic：sh/tcu/VIN001/battery → sh/tcu/VIN001/command
    String topic = record.getProps().get("firstTopic")
                 + "/" + record.getProps().get("secondTopic");
    String commandTopic = topic.substring(0, topic.lastIndexOf('/') + 1) + "command";
    String url = String.format("https://%s/topics/%s?qos=1",
        host, URLEncoder.encode(commandTopic));
    Command cmd = new Command();
    cmd.setAcPower(targetAcPower);
    // HTTP POST 发送（细节略）
}

client.on('message', (topic, message) => {
    if (topic.endsWith('command')) return;  // 跳过控制指令
    const battery = JSON.parse(new TextDecoder().decode(message));
    setLinearProgress(battery.battery);     // 更新电量进度条
});

[00:01] INFO  TCU Emulator started, VIN=VIN001, battery=100%
[00:02] INFO  Battery: 98% → published to sh/tcu/VIN001/battery
[00:12] INFO  Battery: 48% → published
[00:18] WARN  Battery: 28% ≤ 30% threshold
[00:18] SCF   savePower() triggered → ac_power: 2 → 1
[00:18] OK    Command published: {"ac_power":1} → sh/tcu/VIN001/command
[00:19] INFO  TCU received command: ac_power=1, 空调已降档
[00:25] ALERT Battery: 8% ≤ 10% threshold
[00:25] SCF   turnOffAc() triggered → ac_power: 1 → 0
[00:25] OK    Command published: {"ac_power":0} → sh/tcu/VIN001/command
[00:26] INFO  TCU received command: ac_power=0, 空调已关闭