nongye/STM32F103_App1/HardWare/WIFI/mqtt_client.c

/* mqtt_client.c - <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD> MQTT 3.1.1 <20>ͻ<EFBFBD><CDBB>ˣ<EFBFBD>ESP-01S TCP ͸<><CDB8>֮<EFBFBD>ϣ<EFBFBD>
 *
 * ֻ<EFBFBD><EFBFBD><EFBFBD>豸<EFBFBD><EFBFBD>Ҫ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>С<EFBFBD><EFBFBD><EFBFBD><EFBFBD>CONNECT / SUBSCRIBE(QoS0) / PUBLISH(QoS0) / PINGREQ<EFBFBD><EFBFBD>
 * <EFBFBD>Լ<EFBFBD> CONNACK / SUBACK / PINGRESP / PUBLISH <EFBFBD>Ľ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
 * <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>·<EFBFBD><EFBFBD><EFBFBD> PUBLISH payload(JSON) ֱ<EFBFBD>ӽ<EFBFBD><EFBFBD><EFBFBD> network.c <EFBFBD><EFBFBD> net_dispatch_message()<EFBFBD><EFBFBD>
 */

#include "mqtt_client.h"
#include "esp8266.h"
#include "air780e.h"     /* U2_CopyBuff / bg_delay */
#include "network.h"
#include "log.h"
#include "main.h"
#include "string.h"

#define MQTT_TX_BUF_SIZE   512    /* <20><><EFBFBD><EFBFBD><EFBFBD>ͻ<EFBFBD><CDBB>壺CONNECT/SUB/PUBLISH <20><><EFBFBD><EFBFBD>Լ 350B */
#define MQTT_RX_JSON_SIZE  384    /* <20><><EFBFBD><EFBFBD> PUBLISH payload <20>ݴ<EFBFBD> */

static uint8_t  tx_buf[MQTT_TX_BUF_SIZE];
static char     s_cid[24];
static char     s_user[24];
static char     s_pass[24];
static volatile uint8_t s_connack_ok = 0;
static volatile uint8_t s_suback_ok  = 0;
volatile uint8_t g_mqttc_ping_outstanding = 0;

void mqttc_init(const char *client_id, const char *user, const char *pass)
{
    strncpy(s_cid,  client_id, sizeof(s_cid)  - 1); s_cid[sizeof(s_cid)-1]   = 0;
    strncpy(s_user, user,      sizeof(s_user) - 1); s_user[sizeof(s_user)-1] = 0;
    strncpy(s_pass, pass,      sizeof(s_pass) - 1); s_pass[sizeof(s_pass)-1] = 0;
    s_connack_ok = 0;
    s_suback_ok = 0;
    g_mqttc_ping_outstanding = 0;
}

/*==== ʣ<>೤<EFBFBD>ȱ<EFBFBD><C8B1>루MQTT <20>䳤<EFBFBD><E4B3A4><EFBFBD>룬<EFBFBD><EBA3AC><EFBFBD>ǵı<C7B5><C4B1><EFBFBD> < 16KB<4B><42><EFBFBD><EFBFBD><EFBFBD><EFBFBD> 3 <20>ֽڣ<D6BD>====*/
static uint8_t encode_rl(uint8_t *out, uint32_t len)
{
    uint8_t n = 0;
    do {
        uint8_t b = len % 128;
        len /= 128;
        if (len > 0) b |= 0x80;
        out[n++] = b;
    } while (len > 0);
    return n;
}

/* ׷<><D7B7>һ<EFBFBD><D2BB><EFBFBD><EFBFBD> 2 <20>ֽڳ<D6BD><DAB3><EFBFBD>ǰ׺<C7B0><D7BA> UTF-8 <20>ַ<EFBFBD><D6B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ƫ<EFBFBD><C6AB> */
static uint16_t put_str(uint8_t *buf, uint16_t pos, const char *s)
{
    uint16_t l = strlen(s);
    buf[pos++] = (uint8_t)(l >> 8);
    buf[pos++] = (uint8_t)(l & 0xFF);
    memcpy(buf + pos, s, l);
    return pos + l;
}

/*==== CONNECT<43><54>clean session + username + password ====*/
int mqttc_connect(void)
{
    uint16_t pos, var_pos, rl_pos;
    uint8_t rl_len;

    tx_buf[0] = 0x10;
    var_pos = 1 + 3;  /* Ԥ<><D4A4>ʣ<EFBFBD>೤<EFBFBD><E0B3A4>(<28><><EFBFBD><EFBFBD>3<EFBFBD>ֽ<EFBFBD>)<29><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> */

    /* <20>ɱ<EFBFBD>ͷ: "MQTT" + Э<>鼶<EFBFBD><E9BCB6>4 + <20><><EFBFBD>ӱ<EFBFBD>־0xC2 + keepalive */
    var_pos = put_str(tx_buf, var_pos, "MQTT");
    tx_buf[var_pos++] = 0x04;
    tx_buf[var_pos++] = 0xC2;
    tx_buf[var_pos++] = (uint8_t)(MQTT_KEEPALIVE_S >> 8);
    tx_buf[var_pos++] = (uint8_t)(MQTT_KEEPALIVE_S & 0xFF);

    /* <20>غ<EFBFBD>: ClientID / Username / Password */
    var_pos = put_str(tx_buf, var_pos, s_cid);
    var_pos = put_str(tx_buf, var_pos, s_user);
    var_pos = put_str(tx_buf, var_pos, s_pass);

    /* <20><><EFBFBD><EFBFBD>ʣ<EFBFBD>೤<EFBFBD>ȣ<EFBFBD><C8A3>ѿɱ<D1BF>ͷ<EFBFBD><CDB7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ƶ<EFBFBD><C6B6>뵽<EFBFBD><EBB5BD><EFBFBD>볤<EFBFBD>ȣ<EFBFBD> */
    pos = var_pos - (1 + 3);
    rl_len = encode_rl(tx_buf + 1, pos);
    if (rl_len < 3) {
        memmove(tx_buf + 1 + rl_len, tx_buf + 1 + 3, pos);
        var_pos -= (3 - rl_len);
    }
    rl_pos = var_pos;

    log_info("> WiFi MQTT: CONNECT cid=%s", s_cid);
    esp_send(tx_buf, rl_pos);

    /* <20><> CONNACK<43><4B><EFBFBD><EFBFBD><EFBFBD><EFBFBD> 5s<35><73><EFBFBD>ڼ<EFBFBD><DABC><EFBFBD>Ƭι<C6AC><CEB9> */
    s_connack_ok = 0;
    for (int i = 0; i < 50; i++) {
        bg_delay(100);
        while (mqttc_process() == 1) { }
        if (s_connack_ok) {
            log_info("> WiFi MQTT: CONNACK <20>ɹ<EFBFBD>");
            return 0;
        }
    }
    log_warn("> WiFi MQTT: CONNACK <20><>ʱ");
    return -1;
}

/*==== SUBSCRIBE QoS0<53><30><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʶ<EFBFBD>̶<EFBFBD> 1 ====*/
int mqttc_subscribe(const char *topic)
{
    uint16_t pos = 0;

    tx_buf[pos++] = 0x82;
    /* ʣ<>೤<EFBFBD><E0B3A4> = 2(<28><><EFBFBD><EFBFBD>ʶ) + 2 + topic + 1(QoS) */
    pos += encode_rl(tx_buf + pos, (uint32_t)(2 + 2 + strlen(topic) + 1));
    tx_buf[pos++] = 0x00;
    tx_buf[pos++] = 0x01;
    pos = put_str(tx_buf, pos, topic);
    tx_buf[pos++] = 0x00;

    log_info("> WiFi MQTT: SUB %s", topic);
    esp_send(tx_buf, pos);

    s_suback_ok = 0;
    for (int i = 0; i < 50; i++) {
        bg_delay(100);
        while (mqttc_process() == 1) { }
        if (s_suback_ok) {
            log_info("> WiFi MQTT: SUBACK <20>ɹ<EFBFBD>");
            return 0;
        }
    }
    log_warn("> WiFi MQTT: SUBACK <20><>ʱ");
    return -1;
}

/*==== PUBLISH QoS0 ====*/
int mqttc_publish(const char *topic, const char *payload)
{
    uint16_t pos = 0;
    uint32_t rl = 2 + strlen(topic) + strlen(payload);

    if (rl + 5 > MQTT_TX_BUF_SIZE) {
        log_warn("> WiFi MQTT: publish <20><><EFBFBD>Ĺ<EFBFBD><C4B9><EFBFBD> (%u)", (unsigned)rl);
        return -1;
    }

    tx_buf[pos++] = 0x30;
    pos += encode_rl(tx_buf + pos, rl);
    pos = put_str(tx_buf, pos, topic);
    memcpy(tx_buf + pos, payload, strlen(payload));
    pos += strlen(payload);

    esp_send(tx_buf, pos);
    return 0;
}

void mqttc_ping(void)
{
    tx_buf[0] = 0xC0;
    tx_buf[1] = 0x00;
    esp_send(tx_buf, 2);
    g_mqttc_ping_outstanding = 1;
}

/*==== <20><><EFBFBD>ս<EFBFBD><D5BD><EFBFBD> ====*/

/* <20><><EFBFBD><EFBFBD><EFBFBD>ư<EFBFBD>ȫ<EFBFBD><C8AB><EFBFBD>Ӵ<EFBFBD><D3B4><EFBFBD><EFBFBD>ң<EFBFBD>͸<EFBFBD><CDB8><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ܺ<EFBFBD> 0x00<30><30><EFBFBD><EFBFBD> PINGRESP 0xD0 0x00<30><30> */
static uint8_t *memfind(uint8_t *hay, uint16_t hay_len, const char *needle)
{
    uint16_t nlen = strlen(needle);
    if (hay_len < nlen) return NULL;
    for (uint16_t i = 0; i <= hay_len - nlen; i++) {
        if (memcmp(hay + i, needle, nlen) == 0) return hay + i;
    }
    return NULL;
}

/* <20><><EFBFBD>ѻ<EFBFBD><D1BB><EFBFBD>ǰ n <20>ֽ<EFBFBD> */
static void consume(uint16_t n)
{
    if (n >= U2_CopyIndex) {
        U2_CopyIndex = 0;
    } else {
        memmove(U2_CopyBuff, U2_CopyBuff + n, U2_CopyIndex - n);
        U2_CopyIndex -= n;
    }
    U2_CopyBuff[U2_CopyIndex] = '\0';
}

int mqttc_process(void)
{
    uint8_t *buf = U2_CopyBuff;
    uint16_t len = U2_CopyIndex;

    if (len == 0) {
        U2_CopyFlag = 0;
        return 0;
    }

    /* <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>⣺͸<E2A3BA><CDB8>ģʽ<C4A3><CABD> TCP <20>Ͽ<EFBFBD> ESP <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD> "CLOSED" */
    if (memfind(buf, len, "CLOSED")) {
        consume(U2_CopyIndex);
        U2_CopyFlag = 0;
        return -1;
    }

    /* ֡ͬ<D6A1><CDAC><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ֽڸ<D6BD> 4 λ<><CEBB><EFBFBD><EFBFBD><EFBFBD>ǺϷ<C7BA><CFB7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>(1~14)<29><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ֽڶ<D6BD><DAB6><EFBFBD> */
    uint8_t type = buf[0] >> 4;
    if (type < 1 || type > 14) {
        consume(1);
        return 1;
    }

    /* <20><><EFBFBD><EFBFBD>ʣ<EFBFBD>೤<EFBFBD>ȣ<EFBFBD><C8A3><EFBFBD><EFBFBD><EFBFBD> 4 <20>ֽڣ<D6BD> */
    uint32_t rl = 0, mult = 1;
    uint8_t rl_bytes = 0;
    for (uint8_t i = 1; i <= 4; i++) {
        if (i >= len) return 0;  /* ͷ<><CDB7>û<EFBFBD><C3BB>ȫ<EFBFBD><C8AB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> */
        rl += (uint32_t)(buf[i] & 0x7F) * mult;
        rl_bytes = i;
        if (!(buf[i] & 0x80)) break;
        mult *= 128;
    }
    if (rl_bytes == 0) return 0;

    uint32_t total = 1 + rl_bytes + rl;
    if (total > U2_COPY_SIZE - 1) {
        /* <20>쳣<EFBFBD><ECB3A3><EFBFBD>ͱ<EFBFBD><CDB1>ģ<EFBFBD><C4A3><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ֽ<EFBFBD><D6BD><EFBFBD><EFBFBD><EFBFBD>ͬ<EFBFBD><CDAC> */
        consume(1);
        return 1;
    }
    if (len < total) return 0;   /* <20><><EFBFBD><EFBFBD>û<EFBFBD><C3BB>ȫ<EFBFBD><C8AB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> */

    uint8_t *pkt = buf + 1 + rl_bytes;

    switch (type) {
    case 2:  /* CONNACK: 0x20 0x02 0x00 rc */
        if (rl >= 2 && pkt[1] == 0) s_connack_ok = 1;
        else log_warn("> WiFi MQTT: CONNACK <20><><EFBFBD>ܾ<EFBFBD> rc=%u", rl >= 2 ? pkt[1] : 0xFF);
        break;
    case 9:  /* SUBACK: 0x90 0x03 pktIdHi pktIdLo rc<72><63>rc=0x80 <20><>ʾ<EFBFBD><CABE><EFBFBD>ı<EFBFBD><C4B1>ܾ<EFBFBD> */
        if (rl >= 3 && pkt[2] != 0x80) s_suback_ok = 1;
        else log_warn("> WiFi MQTT: SUBACK <20><><EFBFBD>ܾ<EFBFBD>");
        break;
    case 13: /* PINGRESP */
        g_mqttc_ping_outstanding = 0;
        break;
    case 3: { /* PUBLISH QoS0<53><30>ȡ payload(JSON) <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ַ<EFBFBD> */
        static char json[MQTT_RX_JSON_SIZE];
        uint16_t tlen = ((uint16_t)pkt[0] << 8) | pkt[1];
        if (2 + tlen <= rl) {
            uint32_t plen = rl - 2 - tlen;
            if (plen >= sizeof(json)) plen = sizeof(json) - 1;
            memcpy(json, pkt + 2 + tlen, plen);
            json[plen] = '\0';
            log_info("> WiFi MQTT <20><><EFBFBD><EFBFBD>: %.200s", json);
            net_dispatch_message(json);
        }
        break;
    }
    default: /* PUBACK <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ģ<EFBFBD><C4A3><EFBFBD><EFBFBD><EFBFBD> */
        break;
    }

    consume((uint16_t)total);
    return 1;
}