MQTT断线后怎样恢复状态？遗嘱、保留消息和命令去重

// ---- main/reliability.h ----
/* SPDX-License-Identifier: MIT */
#pragma once
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include <inttypes.h>
typedef struct {uint32_t boot, sequence; bool on;} Command;
/* Canonical fixed-width uppercase hex boot, decimal sequence 1..UINT32_MAX, exact ON/OFF. */
static inline bool command_parse(const char *p,size_t n,Command *out) {
 if(!p||!out||n<13||n>24||memchr(p,0,n))return false;
 char text[25];memcpy(text,p,n);text[n]=0;
 uint32_t boot=0;for(unsigned i=0;i<8;i++){char c=text[i];unsigned v=c>='0'&&c<='9'?c-'0':c>='A'&&c<='F'?c-'A'+10:16;if(v>15)return false;boot=(boot<<4)|v;}
 if(text[8]!=':')return false;
 uint64_t seq=0;size_t i=9,start=i;while(i<n&&text[i]>='0'&&text[i]<='9'){seq=seq*10+(text[i++]-'0');if(seq>UINT32_MAX)return false;}
 if(i==start||text[start]=='0'||!seq||i>=n||text[i++]!=':')return false;
 bool on=!strcmp(text+i,"ON");if(!on&&strcmp(text+i,"OFF"))return false;
 *out=(Command){boot,(uint32_t)seq,on};return true;
}
static inline const char *command_reason(Command c,uint32_t boot,uint32_t last,bool retained){
 if(retained)return "RETAINED_REJECTED";
 if(c.boot!=boot)return "WRONG_BOOT";
 if(c.sequence<=last)return "DUPLICATE_OR_OLD";
 return "APPLY";
}

// ---- main/main.c ----
/* SPDX-License-Identifier: MIT */
#include <stdio.h>
#include <string.h>
#include <inttypes.h>
#include <stdatomic.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/queue.h"
#include "driver/gpio.h"
#include "esp_log.h"
#include "esp_random.h"
#include "esp_mac.h"
#include "esp_timer.h"
#include "mqtt_client.h"
#include "sdkconfig.h"
#include "net.h"
#include "reliability.h"
#include "sensirion_hal_idf.h"
#include "sensirion_i2c_hal.h"
#include "sht4x_i2c.h"
static const char *TAG="N07";
#define BASE "hixinwu/n07"
static QueueHandle_t commands;
static atomic_bool connected, state_dirty;
static uint32_t boot, applied; /* app_main owns applied and output; callback never changes them */
static bool led;
static void event(void *arg,esp_event_base_t base,int32_t id,void *data) {
 (void)arg;(void)base;esp_mqtt_event_handle_t e=data;
 if(id==MQTT_EVENT_CONNECTED){
  atomic_store(&connected,true);atomic_store(&state_dirty,true);
  int sub=esp_mqtt_client_subscribe(e->client,BASE "/command",1);
  int pub=esp_mqtt_client_enqueue(e->client,BASE "/availability","online",0,1,1,true);
  ESP_LOGI(TAG,"connected subscribe_id=%d online_id=%d; not an application acknowledgement",sub,pub);
 }else if(id==MQTT_EVENT_DISCONNECTED){atomic_store(&connected,false);ESP_LOGW(TAG,"offline; SDK will reconnect");}
 else if(id==MQTT_EVENT_DATA){
  bool topic=e->topic&&e->topic_len==strlen(BASE "/command")&&!memcmp(e->topic,BASE "/command",e->topic_len);
  Command c;bool whole=e->current_data_offset==0&&e->data_len==e->total_data_len;
  if(!topic||!whole||e->retain||!command_parse(e->data,e->data_len,&c)){ESP_LOGW(TAG,"reject topic/fragment/retained/syntax");return;}
  if(xQueueSend(commands,&c,0)!=pdTRUE)ESP_LOGW(TAG,"COMMAND_QUEUE_FULL: not applied; sender must inspect applied sequence");
 }else if(id==MQTT_EVENT_PUBLISHED)ESP_LOGI(TAG,"PUBACK id=%d (broker hop, not physical LED proof)",e->msg_id);
 else if(id==MQTT_EVENT_ERROR)ESP_LOGW(TAG,"MQTT error; no credentials logged");
}
static int state(esp_mqtt_client_handle_t c) {
 char p[160];snprintf(p,sizeof(p),"{\"boot\":\"%08" PRIX32 "\",\"applied_sequence\":%" PRIu32 ",\"gpio4\":%s}",boot,applied,led?"true":"false");
 return esp_mqtt_client_enqueue(c,BASE "/state",p,0,1,1,true);
}
void app_main(void) {
 boot=esp_random();commands=xQueueCreate(4,sizeof(Command));if(!commands){ESP_LOGE(TAG,"queue allocation failed");return;}
 gpio_reset_pin(GPIO_NUM_4);gpio_set_direction(GPIO_NUM_4,GPIO_MODE_OUTPUT);gpio_set_level(GPIO_NUM_4,0);
 net_start();sensirion_i2c_hal_init();sht4x_init(0x44);
 uint8_t mac[6];ESP_ERROR_CHECK(esp_read_mac(mac,ESP_MAC_WIFI_STA));static char client_id[40];snprintf(client_id,sizeof(client_id),"hixinwu-n07-%02X%02X%02X%02X%02X%02X",mac[0],mac[1],mac[2],mac[3],mac[4],mac[5]);
 esp_mqtt_client_config_t cfg={.broker.address.uri=CONFIG_DEMO_BROKER_URL,.credentials.client_id=client_id,
 .session={.keepalive=10,.message_retransmit_timeout=5000,.last_will={.topic=BASE "/availability",.msg="offline",.qos=1,.retain=1}},
 .network={.timeout_ms=3000,.reconnect_timeout_ms=3000},.buffer={.size=512,.out_size=512},.outbox.limit=16384};
 esp_mqtt_client_handle_t client=esp_mqtt_client_init(&cfg);if(!client){ESP_LOGE(TAG,"MQTT allocation failed");return;}
 ESP_ERROR_CHECK(esp_mqtt_client_register_event(client,ESP_EVENT_ANY_ID,event,NULL));
 bool started=false;uint32_t sample_sequence=0;int64_t next_sample=0;
 for(;;){
  if(!started&&net_online()){ESP_ERROR_CHECK(esp_mqtt_client_start(client));started=true;}
  Command c;while(xQueueReceive(commands,&c,0)==pdTRUE){
   const char *why=command_reason(c,boot,applied,false);
   if(!strcmp(why,"APPLY")){led=c.on;gpio_set_level(GPIO_NUM_4,led);applied=c.sequence;}
   ESP_LOGI(TAG,"command seq=%" PRIu32 " result=%s; digital write, inspect actual LED separately",c.sequence,why);
   atomic_store(&state_dirty,true);
  }
  if(atomic_load(&connected)&&atomic_exchange(&state_dirty,false)){
   if(state(client)<0){atomic_store(&state_dirty,true);ESP_LOGW(TAG,"STATE_ENQUEUE_FAILED: retry latest snapshot");}
  }
  int64_t now=esp_timer_get_time()/1000;
  if(now>=next_sample){
   next_sample=now+5000;atomic_store(&state_dirty,true);int32_t t=0,rh=0;int16_t err=demo_i2c_ready()?sht4x_measure_high_precision(&t,&rh):-1;
   char p[256];sample_sequence++;
   if(!err){int32_t clipped=rh<0?0:rh>100000?100000:rh;
    snprintf(p,sizeof(p),"{\"boot\":\"%08" PRIX32 "\",\"sequence\":%" PRIu32 ",\"valid\":true,\"temperature_c\":%.3f,\"humidity_percent\":%.3f}",boot,sample_sequence,t/1000.0,clipped/1000.0);
   }else snprintf(p,sizeof(p),"{\"boot\":\"%08" PRIX32 "\",\"sequence\":%" PRIu32 ",\"valid\":false,\"temperature_c\":null,\"humidity_percent\":null}",boot,sample_sequence);
   int msg=atomic_load(&connected)?esp_mqtt_client_enqueue(client,BASE "/telemetry",p,0,1,0,true):-1;
   ESP_LOGI(TAG,"sample seq=%" PRIu32 " sensor=%d enqueue=%d outbox=%d; no app backfill",sample_sequence,err,msg,esp_mqtt_client_get_outbox_size(client));
  }
  vTaskDelay(pdMS_TO_TICKS(50));
 }
}