S3：网关 + 多节点——把「脑」和「手脚」接起来

#include "mqtt_client.h"
#include "esp_log.h"
#include <string.h>

static const char *TAG = "node";

/* 节点自己的身份，烧录时改这三个宏 */
#define NODE_ID      "node01"
#define WIFI_SSID    "your-ssid"
#define WIFI_PASS    "your-pass"

static esp_mqtt_client_handle_t s_client = NULL;
static volatile bool s_online = false;

static void mqtt_event_handler(void *args, esp_event_base_t base,
                               int32_t event_id, void *event_data)
{
    esp_mqtt_event_handle_t ev = (esp_mqtt_event_handle_t)event_data;

    switch ((esp_mqtt_event_id_t)event_id) {
    case MQTT_EVENT_CONNECTED: {
        s_online = true;
        ESP_LOGI(TAG, "connected");

        /* 连上先报「我在线」 */
        esp_mqtt_client_publish(s_client,
            "home/" NODE_ID "/status", "online", 0, 1, /*retain=*/1);

        /* 订阅自己的命令主题 */
        esp_mqtt_client_subscribe(s_client, "home/" NODE_ID "/cmd/+", 1);

        /* 把断网期间攒的数据补发出去 */
        extern void cache_flush(void);
        cache_flush();
        break;
    }
    case MQTT_EVENT_DISCONNECTED:
        s_online = false;
        ESP_LOGW(TAG, "disconnected, will retry");
        break;

    case MQTT_EVENT_DATA:
        /* 主题可能很长，不能当 C 字符串直接用 —— 见 4.4 */
        handle_command(ev->topic, ev->topic_len, ev->data, ev->data_len);
        break;

    default: break;
    }
}

static void mqtt_start(void)
{
    esp_mqtt_client_config_t cfg = {
        .broker.address.uri = "mqtt://192.168.1.10:1883",
        .credentials.client_id = NODE_ID,

        /* keepalive：多久没心跳就判定掉线（秒） */
        .session.keepalive = 15,

        /* 遗嘱：异常断开时，Broker 代我发这条 */
        .session.last_will.topic  = "home/" NODE_ID "/status",
        .session.last_will.msg    = "offline",
        .session.last_will.msg_len = 7,
        .session.last_will.qos    = 1,
        .session.last_will.retain = 1,
    };

    s_client = esp_mqtt_client_init(&cfg);
    esp_mqtt_client_register_event(s_client, ESP_EVENT_ANY_ID, mqtt_event_handler, NULL);
    esp_mqtt_client_start(s_client);
}

#define CACHE_MAX 60      /* 60 条 × 每 30 秒一条 = 30 分钟 */

typedef struct {
    uint32_t ts;          /* ⚠️ 必须是「采集时刻」，不是「发送时刻」 */
    float    temp;
    float    rh;
} sample_t;

static sample_t  s_cache[CACHE_MAX];
static int       s_head = 0;    /* 写指针 */
static int       s_tail = 0;    /* 读指针 */
static int       s_cnt  = 0;

/* 生产者：无论在线离线，采集后一律先入缓存 */
static void cache_push(const sample_t *s)
{
    s_cache[s_head] = *s;
    s_head = (s_head + 1) % CACHE_MAX;
    if (s_cnt < CACHE_MAX) {
        s_cnt++;
    } else {
        /* 满了：覆盖最旧，读指针跟进 —— 丢最旧的，保最新的 */
        s_tail = (s_tail + 1) % CACHE_MAX;
    }
}

/* 消费者：只在在线时调用 */
static bool cache_pop(sample_t *out)
{
    if (s_cnt == 0) return false;
    *out = s_cache[s_tail];
    s_tail = (s_tail + 1) % CACHE_MAX;
    s_cnt--;
    return true;
}

void cache_flush(void)
{
    sample_t s;
    int flushed = 0;
    while (cache_pop(&s)) {
        char payload[96];
        /* 关键：把原始时间戳一起带上 */
        snprintf(payload, sizeof(payload),
                 "{\"ts\":%lu,\"t\":%.1f,\"rh\":%.1f}", (unsigned long)s.ts, s.temp, s.rh);
        esp_mqtt_client_publish(s_client,
            "home/" NODE_ID "/sensor/bundle", payload, 0, 1, 0);
        flushed++;
        if (flushed >= 10) break;   /* 一次别补太多，给别的任务留机会 */
    }
    if (flushed) ESP_LOGI(TAG, "flushed %d cached samples", flushed);
}

/* ❌ 错误：topic 不是以 \0 结尾的 */
if (strcmp(ev->topic, "home/node01/cmd/relay") == 0) { ... }

/* ✅ 正确：用 topic_len 限定 */
static bool topic_is(esp_mqtt_event_handle_t ev, const char *want)
{
    size_t n = strlen(want);
    return (ev->topic_len == (int)n) && (memcmp(ev->topic, want, n) == 0);
}

/* 自己控制重连节奏（关闭库自动重连） */
.cfg.network.disable_auto_reconnect = true,

static void reconnect_task(void *arg)
{
    int backoff_ms = 1000;
    const int cap_ms = 30000;

    for (;;) {
        if (!s_online) {
            esp_mqtt_client_reconnect(s_client);
            vTaskDelay(pdMS_TO_TICKS(backoff_ms));

            if (!s_online) {
                /* 指数退避 + ±20% 抖动（避免多节点同步重连） */
                int jitter = (backoff_ms / 5) * ((esp_random() % 3) - 1);
                backoff_ms = backoff_ms * 2 + jitter;
                if (backoff_ms > cap_ms) backoff_ms = cap_ms;
            } else {
                backoff_ms = 1000;   /* 连上了，重置 */
            }
        } else {
            vTaskDelay(pdMS_TO_TICKS(1000));
        }
    }
}

sudo apt install -y mosquitto mosquitto-clients libmosquitto-dev
sudo systemctl enable --now mosquitto

# 验证 Broker 起来了
mosquitto_sub -h localhost -t 'test' -v &
mosquitto_pub -h localhost -t 'test' -m 'hello'

#include <mosquitto.h>
#include <stdio.h>
#include <string.h>
#include <time.h>

#define BROKER_HOST "127.0.0.1"
#define BROKER_PORT 1883
#define GW_ID       "gw-collector"

/* 收到消息：把主题拆开，取出 <id> 和 <metric> */
static void on_message(struct mosquitto *m, void *obj,
                       const struct mosquitto_message *msg)
{
    if (!msg->payload || msg->payloadlen <= 0) return;
    if (!msg->topic) return;

    /* 主题形如 home/node01/sensor/temp
       用 sscanf 精确分段，别用 strtok（它会改原串） */
    char dev[32] = {0}, kind[16] = {0}, metric[32] = {0};
    if (sscanf(msg->topic, "home/%31[^/]/%15[^/]/%31[^/]", dev, kind, metric) != 3) {
        return;   /* 层级不对，不是我们的主题 */
    }

    time_t now = time(NULL);
    char *payload = strndup((const char *)msg->payload, msg->payloadlen);
    if (!payload) return;

    if (strcmp(kind, "sensor") == 0) {
        printf("[%ld] %s.%s = %s\n", (long)now, dev, metric, payload);
        /* 这里写数据库 / 内存表 / 环形缓冲，供网页读 */
    } else if (strcmp(kind, "status") == 0) {
        printf("[%ld] %s is %s\n", (long)now, dev, payload);
        /* 更新在线表：online / offline */
    }

    free(payload);
}

int main(void)
{
    mosquitto_lib_init();

    struct mosquitto *m = mosquitto_new(GW_ID, true, NULL);
    if (!m) { fprintf(stderr, "new failed\n"); return 1; }

    mosquitto_message_callback_set(m, on_message);

    /* keepalive 20s：比节点端 15s 大一点，避免网关先判死 */
    if (mosquitto_connect(m, BROKER_HOST, BROKER_PORT, 20) != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "connect failed\n");
        return 1;
    }

    /* 订阅全部节点数据 + 状态。注意最后一个参数是 QoS */
    mosquitto_subscribe(m, NULL, "home/+/sensor/#", 1);
    mosquitto_subscribe(m, NULL, "home/+/status",   1);

    /* loop_forever 内部处理重连，不用自己写 */
    mosquitto_loop_forever(m, -1, 1);

    mosquitto_destroy(m);
    mosquitto_lib_cleanup();
    return 0;
}

gcc -std=c11 -Wall -Wextra -Werror -O2 \
    -o gw-collector gw-collector.c -lmosquitto

/* 网页点按钮 → 网关发一条命令 */
int send_relay_cmd(const char *node_id, int on)
{
    char topic[96], payload[8];
    snprintf(topic, sizeof(topic), "home/%s/cmd/relay", node_id);
    snprintf(payload, sizeof(payload), "%d", on ? 1 : 0);

    /* QoS 1：命令不能丢。retain 0：命令是「事件」，不是「状态」 */
    int rc = mosquitto_publish(s_client, NULL, topic,
                               (int)strlen(payload), payload, 1, 0);
    return rc == MOSQ_ERR_SUCCESS ? 0 : -1;
}

# 网关跑起来后，另开一个终端，模拟一个节点发数据
mosquitto_pub -h localhost -t 'home/node99/sensor/temp' -m '25.3'

# 看网关输出里有没有 [时间] node99.temp = 25.3

/* 节点只发值，不发时间 */
esp_mqtt_client_publish(s_client, "home/node01/sensor/temp", "25.3", 0, 1, 0);

/* 网关收到时用本地时间 */
time_t ts = time(NULL);

/* 节点订阅一个对时主题，网关定期广播 */
esp_mqtt_client_subscribe(s_client, "home/broadcast/time", 0);

/* 收到后设置系统时间 */
case MQTT_EVENT_DATA:
    if (topic_is(ev, "home/broadcast/time")) {
        time_t t = (time_t)strtoll(payload, NULL, 10);
        struct timeval tv = { .tv_sec = t };
        settimeofday(&tv, NULL);
        ESP_LOGI(TAG, "time synced: %ld", (long)t);
    }

char buf[24];
snprintf(buf, sizeof(buf), "%ld", (long)time(NULL));
mosquitto_publish(m, NULL, "home/broadcast/time", strlen(buf), buf, 0, 0);

{"ts":1730611200,"t":25.3,"rh":58.1}

/* 拼：定长格式，注意 %lu 和 %.1f */
snprintf(buf, sizeof(buf), "{\"ts\":%lu,\"t\":%.1f,\"rh\":%.1f}",
         (unsigned long)ts, temp, rh);

/* 拆：字段顺序固定，直接按格式读 */
unsigned long ts; float t, rh;
if (sscanf(payload, "{\"ts\":%lu,\"t\":%f,\"rh\":%f}", &ts, &t, &rh) == 3) { ... }