小智机器人关键函数解析：MqttProtocol::OpenAudioChannel()通过MQTT协议打开一个音频通道，从服务器接收加密音频数据，解密后通过回调函数处理音频数据

LS_learner

545人浏览 · 2025-03-31 00:37:05

LS_learner · 2025-03-31 00:37:05 发布

这段代码定义了 MqttProtocol 类中的 OpenAudioChannel 方法，其主要功能是通过MQTT协议打开一个音频通道，从服务器接收加密音频数据，解密后通过回调函数处理音频数据。
源码：

bool MqttProtocol::OpenAudioChannel() {
    if (mqtt_ == nullptr || !mqtt_->IsConnected()) {
        ESP_LOGI(TAG, "MQTT is not connected, try to connect now");
        if (!StartMqttClient()) {
            return false;
        }
    }

    session_id_ = "";
    xEventGroupClearBits(event_group_handle_, MQTT_PROTOCOL_SERVER_HELLO_EVENT);

    // 发送 hello 消息申请 UDP 通道
    std::string message = "{";
    message += "\"type\":\"hello\",";
    message += "\"version\": 3,";
    message += "\"transport\":\"udp\",";
    message += "\"audio_params\":{";
    message += "\"format\":\"opus\", \"sample_rate\":16000, \"channels\":1, \"frame_duration\":" + std::to_string(OPUS_FRAME_DURATION_MS);
    message += "}}";
    SendText(message);

    // 等待服务器响应
    EventBits_t bits = xEventGroupWaitBits(event_group_handle_, MQTT_PROTOCOL_SERVER_HELLO_EVENT, pdTRUE, pdFALSE, pdMS_TO_TICKS(10000));
    if (!(bits & MQTT_PROTOCOL_SERVER_HELLO_EVENT)) {
        ESP_LOGE(TAG, "Failed to receive server hello");
        if (on_network_error_ != nullptr) {
            on_network_error_(Lang::Strings::WAITING_FOR_RESPONSE_TIMEOUT);
        }
        return false;
    }

    std::lock_guard<std::mutex> lock(channel_mutex_);
    if (udp_ != nullptr) {
        delete udp_;
    }
    udp_ = Board::GetInstance().CreateUdp();
    udp_->OnMessage([this](const std::string& data) {
        if (data.size() < sizeof(aes_nonce_)) {
            ESP_LOGE(TAG, "Invalid audio packet size: %zu", data.size());
            return;
        }
        if (data[0] != 0x01) {
            ESP_LOGE(TAG, "Invalid audio packet type: %x", data[0]);
            return;
        }
        uint32_t sequence = ntohl(*(uint32_t*)&data[12]);
        if (sequence < remote_sequence_) {
            //ESP_LOGW(TAG, "Received audio packet with old sequence: %lu, expected: %lu", sequence, remote_sequence_);
            return;
        }
        if (sequence != remote_sequence_ + 1) {
            //ESP_LOGW(TAG, "Received audio packet with wrong sequence: %lu, expected: %lu", sequence, remote_sequence_ + 1);
        }

        std::vector<uint8_t> decrypted;
        size_t decrypted_size = data.size() - aes_nonce_.size();
        size_t nc_off = 0;
        uint8_t stream_block[16] = {0};
        decrypted.resize(decrypted_size);
        auto nonce = (uint8_t*)data.data();
        auto encrypted = (uint8_t*)data.data() + aes_nonce_.size();
        int ret = mbedtls_aes_crypt_ctr(&aes_ctx_, decrypted_size, &nc_off, nonce, stream_block, encrypted, (uint8_t*)decrypted.data());
        if (ret != 0) {
            ESP_LOGE(TAG, "Failed to decrypt audio data, ret: %d", ret);
            return;
        }
        if (on_incoming_audio_ != nullptr) {
            on_incoming_audio_(std::move(decrypted));
        }
        remote_sequence_ = sequence;
    });

    udp_->Connect(udp_server_, udp_port_);

    if (on_audio_channel_opened_ != nullptr) {
        on_audio_channel_opened_();
    }
    return true;
}

这段代码定义了 MqttProtocol 类中的 OpenAudioChannel 方法，其主要功能是通过MQTT协议打开一个音频通道，从服务器接收加密音频数据，解密后通过回调函数处理音频数据。以下是对代码的详细解释：

函数头和参数检查

bool MqttProtocol::OpenAudioChannel() {
    if (mqtt_ == nullptr || !mqtt_->IsConnected()) {
        ESP_LOGI(TAG, "MQTT is not connected, try to connect now");
        if (!StartMqttClient()) {
            return false;
        }
    }

MqttProtocol::OpenAudioChannel(): 这是 MqttProtocol 类的成员函数，用于打开音频通道。
检查 mqtt_ 是否为空或未连接。如果未连接，尝试调用 StartMqttClient() 函数连接MQTT服务器。若连接失败，返回 false。

初始化会话和事件标志

    session_id_ = "";
    xEventGroupClearBits(event_group_handle_, MQTT_PROTOCOL_SERVER_HELLO_EVENT);

session_id_ = "": 清空会话ID。
xEventGroupClearBits: 清除事件组中的 MQTT_PROTOCOL_SERVER_HELLO_EVENT 标志，用于后续等待服务器响应。

发送Hello消息

    // 发送 hello 消息申请 UDP 通道
    std::string message = "{";
    message += "\"type\":\"hello\",";
    message += "\"version\": 3,";
    message += "\"transport\":\"udp\",";
    message += "\"audio_params\":{";
    message += "\"format\":\"opus\", \"sample_rate\":16000, \"channels\":1, \"frame_duration\":" + std::to_string(OPUS_FRAME_DURATION_MS);
    message += "}}";
    SendText(message);

构建一个JSON格式的Hello消息，指定音频传输使用UDP协议、音频格式为Opus、采样率为16000Hz、单声道等参数。
调用 SendText 函数将消息发送到MQTT服务器。

等待服务器响应

    // 等待服务器响应
    EventBits_t bits = xEventGroupWaitBits(event_group_handle_, MQTT_PROTOCOL_SERVER_HELLO_EVENT, pdTRUE, pdFALSE, pdMS_TO_TICKS(10000));
    if (!(bits & MQTT_PROTOCOL_SERVER_HELLO_EVENT)) {
        ESP_LOGE(TAG, "Failed to receive server hello");
        if (on_network_error_ != nullptr) {
            on_network_error_(Lang::Strings::WAITING_FOR_RESPONSE_TIMEOUT);
        }
        return false;
    }

使用 xEventGroupWaitBits 函数等待 MQTT_PROTOCOL_SERVER_HELLO_EVENT 标志被设置，最多等待10秒。
如果未收到服务器响应，记录错误日志，调用 on_network_error_ 回调函数，返回 false。

创建并配置UDP客户端

    std::lock_guard<std::mutex> lock(channel_mutex_);
    if (udp_ != nullptr) {
        delete udp_;
    }
    udp_ = Board::GetInstance().CreateUdp();
    udp_->OnMessage([this](const std::string& data) {
        // 处理接收到的音频数据
        // ...
    });

使用 std::lock_guard 加锁，确保线程安全。
如果 udp_ 不为空，释放之前的UDP客户端。
通过 Board::GetInstance().CreateUdp() 创建一个新的UDP客户端。
使用 udp_->OnMessage 设置回调函数，处理接收到的音频数据。

处理接收到的音频数据

        if (data.size() < sizeof(aes_nonce_)) {
            ESP_LOGE(TAG, "Invalid audio packet size: %zu", data.size());
            return;
        }
        if (data[0] != 0x01) {
            ESP_LOGE(TAG, "Invalid audio packet type: %x", data[0]);
            return;
        }
        uint32_t sequence = ntohl(*(uint32_t*)&data[12]);
        if (sequence < remote_sequence_) {
            //ESP_LOGW(TAG, "Received audio packet with old sequence: %lu, expected: %lu", sequence, remote_sequence_);
            return;
        }
        if (sequence != remote_sequence_ + 1) {
            //ESP_LOGW(TAG, "Received audio packet with wrong sequence: %lu, expected: %lu", sequence, remote_sequence_ + 1);
        }

检查音频数据包的大小和类型是否合法。
提取数据包的序列号，并检查是否为预期的序列号。

解密音频数据

        std::vector<uint8_t> decrypted;
        size_t decrypted_size = data.size() - aes_nonce_.size();
        size_t nc_off = 0;
        uint8_t stream_block[16] = {0};
        decrypted.resize(decrypted_size);
        auto nonce = (uint8_t*)data.data();
        auto encrypted = (uint8_t*)data.data() + aes_nonce_.size();
        int ret = mbedtls_aes_crypt_ctr(&aes_ctx_, decrypted_size, &nc_off, nonce, stream_block, encrypted, (uint8_t*)decrypted.data());
        if (ret != 0) {
            ESP_LOGE(TAG, "Failed to decrypt audio data, ret: %d", ret);
            return;
        }

分配内存用于存储解密后的音频数据。
使用 mbedtls_aes_crypt_ctr 函数对音频数据进行解密。
如果解密失败，记录错误日志并返回。

处理解密后的音频数据

        if (on_incoming_audio_ != nullptr) {
            on_incoming_audio_(std::move(decrypted));
        }
        remote_sequence_ = sequence;

如果 on_incoming_audio_ 回调函数不为空，调用该函数处理解密后的音频数据。
更新远程序列号。

连接UDP服务器并回调

    udp_->Connect(udp_server_, udp_port_);

    if (on_audio_channel_opened_ != nullptr) {
        on_audio_channel_opened_();
    }
    return true;

调用 udp_->Connect 连接到UDP服务器。
如果 on_audio_channel_opened_ 回调函数不为空，调用该函数表示音频通道已打开。
返回 true 表示音频通道打开成功。

DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

2026 年 AI 前沿技术全行业应用场景汇总

DAMO开发者矩阵

从零实现一个微信机器人：基于 Node.js 与 Webhook 回调的工程化落地指南

在企业私域运营与自动化办公的场景中，搭建一个能够实时响应的自动化消息助手能够极大释放人力。本文将抛开复杂的客户端底层逆向，从标准的视角，带你。

DAMO开发者矩阵

面向高校科研开发的轮式双臂具身机器人研究综述与平台选型

DAMO开发者矩阵

所有评论(0)

查看更多评论

LS_learner

@qq_39777550

已为社区贡献28条内容

小智机器人关键函数解析：MqttProtocol::OpenAudioChannel()通过MQTT协议打开一个音频通道，从服务器接收加密音频数据，解密后通过回调函数处理音频数据

LS_learner

函数头和参数检查

初始化会话和事件标志

发送Hello消息

等待服务器响应

创建并配置UDP客户端

处理接收到的音频数据

解密音频数据

处理解密后的音频数据

连接UDP服务器并回调

所有评论(0)

温馨提示：您尚未绑定手机号

LS_learner