AC7911BB_JL-AIBT-SoundBox/mqtt/mqtt_protocol.c

#include "mqtt_protocol.h"

#include "mqtt_config.h"
#include "mqtt_manager.h"
#include "app_log.h"
#include "system/includes.h"

#include "json_c/json.h"

#include <string.h>
#include <stdio.h>
#include <stdarg.h>


/*
 * 当前会话ID
 */
static char protocol_session_id[64];


/*
 * 协议消息回调
 */
static mqtt_protocol_callback_t protocol_callback = NULL;


/*
 * 解析type字段
 */
static enum mqtt_protocol_message_type parse_type(const char *value)
{

    if (!value)
    {
        return MQTT_PROTOCOL_MSG_UNKNOWN;
    }

    if (!strcmp(value, "hello"))
    {
        return MQTT_PROTOCOL_MSG_HELLO;
    }

    if (!strcmp(value, "tts"))
    {
        return MQTT_PROTOCOL_MSG_TTS;
    }

    if (!strcmp(value, "stt"))
    {
        return MQTT_PROTOCOL_MSG_STT;
    }

    if (!strcmp(value, "llm"))
    {
        return MQTT_PROTOCOL_MSG_LLM;
    }

    if (!strcmp(value, "listen"))
    {
        return MQTT_PROTOCOL_MSG_LISTEN;
    }

    if (!strcmp(value, "goodbye"))
    {
        return MQTT_PROTOCOL_MSG_GOODBYE;
    }

    return MQTT_PROTOCOL_MSG_UNKNOWN;

}


/*
 * 解析state字段
 */
static enum mqtt_protocol_message_state parse_state(const char *value)
{

    if (!value)
    {
        return MQTT_PROTOCOL_STATE_NONE;
    }

    if (!strcmp(value, "start"))
    {
        return MQTT_PROTOCOL_STATE_START;
    }

    if (!strcmp(value, "sentence_start"))
    {
        return MQTT_PROTOCOL_STATE_SENTENCE_START;
    }

    if (!strcmp(value, "stop"))
    {
        return MQTT_PROTOCOL_STATE_STOP;
    }

    if (!strcmp(value, "detect"))
    {
        return MQTT_PROTOCOL_STATE_DETECT;
    }

    return MQTT_PROTOCOL_STATE_NONE;

}


/*
 * 解析收到的MQTT消息
 */
static int protocol_parse
(
    const char *payload,
    uint32_t len,
    struct mqtt_protocol_message *message
)
{

    json_object *root;
    json_object *item;
    char *json_buf;
    const char *value;

    if (!payload || !len || !message)
    {
        return -1;
    }

    memset(message, 0, sizeof(*message));

    json_buf = zalloc(len + 1);
    if (!json_buf)
    {
        return -1;
    }
    memcpy(json_buf, payload, len);

    root = json_tokener_parse(json_buf);
    free(json_buf);

    if (!root)
    {
        return -1;
    }

    if (json_object_object_get_ex(root, "type", &item))
    {
        message->type = parse_type(json_object_get_string(item));
    }

    if (json_object_object_get_ex(root, "state", &item))
    {
        message->state = parse_state(json_object_get_string(item));
    }

    if (json_object_object_get_ex(root, "code", &item))
    {
        message->has_code = 1;
        message->code = json_object_get_int(item);
    }

    if (json_object_object_get_ex(root, "version", &item))
    {
        message->has_version = 1;
        message->version = json_object_get_int(item);
    }

    if (json_object_object_get_ex(root, "transport", &item))
    {
        value = json_object_get_string(item);
        if (value)
        {
            snprintf(message->transport, sizeof(message->transport), "%s", value);
        }
    }

    if (json_object_object_get_ex(root, "session_id", &item))
    {
        value = json_object_get_string(item);
        if (value)
        {
            snprintf(message->session_id, sizeof(message->session_id), "%s", value);
        }
    }

    if (json_object_object_get_ex(root, "text", &item))
    {
        value = json_object_get_string(item);
        if (value)
        {
            snprintf(message->text, sizeof(message->text), "%s", value);
        }
    }

    json_object_put(root);

    return message->type == MQTT_PROTOCOL_MSG_UNKNOWN ? -1 : 0;

}


/*
 * 发送JSON文本到发布topic
 */
static int protocol_send_text(const char *text)
{

    if (!text)
    {
        return -1;
    }

    return mqtt_manager_publish(
        mqtt_config_get()->publish_topic,
        (const uint8_t *)text,
        strlen(text)
    );

}


/*
 * 带session_id拼接JSON
 */
static int protocol_send_jsonf(const char *fmt, ...)
{

    char buf[256];
    va_list ap;
    int len;

    if (!fmt)
    {
        return -1;
    }

    va_start(ap, fmt);
    len = vsnprintf(buf, sizeof(buf), fmt, ap);
    va_end(ap);

    if (len <= 0 || len >= sizeof(buf))
    {
        return -1;
    }

    return protocol_send_text(buf);

}


int mqtt_protocol_handle_message
(
    const char *topic,
    const uint8_t *payload,
    uint32_t len
)
{

    struct mqtt_protocol_message message;

    if (protocol_parse((const char *)payload, len, &message))
    {
        return -1;
    }

    if (message.session_id[0])
    {
        strcpy(protocol_session_id, message.session_id);
    }

    if (message.type == MQTT_PROTOCOL_MSG_HELLO)
    {
        APP_LOG(
            "[MQTT_P] hello: code=%d session=%s\n",
            message.has_code ? message.code : -1,
            protocol_session_id
        );
    }
    else if (message.type == MQTT_PROTOCOL_MSG_GOODBYE)
    {
        APP_LOG(
            "[MQTT_P] goodbye: session=%s\n",
            message.session_id[0] ? message.session_id : "-"
        );
        protocol_session_id[0] = '\0';
    }

    if (protocol_callback)
    {
        protocol_callback(&message);
    }

    return 0;

}


int mqtt_protocol_send_hello(void)
{

    return protocol_send_jsonf(
        "{\"type\":\"hello\",\"version\":3,\"transport\":\"udp\","
        "\"features\":{\"mcp\":true},"
        "\"audio_params\":{\"format\":\"opus\",\"sample_rate\":16000,"
        "\"channels\":1,\"frame_duration\":60}}"
    );

}


int mqtt_protocol_send_listen_start(void)
{

    if (protocol_session_id[0])
    {
        return protocol_send_jsonf(
            "{\"type\":\"listen\",\"state\":\"start\",\"session_id\":\"%s\"}",
            protocol_session_id
        );
    }

    return protocol_send_jsonf("{\"type\":\"listen\",\"state\":\"start\"}");

}


int mqtt_protocol_send_listen_stop(void)
{

    if (protocol_session_id[0])
    {
        return protocol_send_jsonf(
            "{\"type\":\"listen\",\"state\":\"stop\",\"session_id\":\"%s\"}",
            protocol_session_id
        );
    }

    return protocol_send_jsonf("{\"type\":\"listen\",\"state\":\"stop\"}");

}


int mqtt_protocol_send_abort(void)
{

    if (protocol_session_id[0])
    {
        return protocol_send_jsonf(
            "{\"type\":\"abort\",\"session_id\":\"%s\"}",
            protocol_session_id
        );
    }

    return protocol_send_jsonf("{\"type\":\"abort\"}");

}


int mqtt_protocol_send_goodbye(void)
{

    if (protocol_session_id[0])
    {
        return protocol_send_jsonf(
            "{\"type\":\"goodbye\",\"session_id\":\"%s\"}",
            protocol_session_id
        );
    }

    return -1;

}


const char *mqtt_protocol_session_id(void)
{

    return protocol_session_id;

}


void mqtt_protocol_set_callback(mqtt_protocol_callback_t callback)
{

    protocol_callback = callback;

}