add sening functions

wangzhibo

2026-08-26 70481429367ca00144b24f118cbfdfb28b02967d

 xiaozhi-esp32-2.2.6/main/protocols/mqtt_protocol.cc

@@ -18,11 +18,11 @@
        .callback = [](void* arg) {
            MqttProtocol* protocol = (MqttProtocol*)arg;
            auto& app = Application::GetInstance();
            if (app.GetDeviceState() == kDeviceStateIdle) {
            if (app.GetDeviceState() == kDeviceStateIdle && Board::GetInstance().IsNetworkReady()) {
                ESP_LOGI(TAG, "Reconnecting to MQTT server");
                auto alive = protocol->alive_;  // Capture alive flag
                app.Schedule([protocol, alive]() {
                    if (*alive) {
                    if (*alive && Board::GetInstance().IsNetworkReady()) {
                        protocol->StartMqttClient(false);
                    }
                });
@@ -56,9 +56,25 @@
    return StartMqttClient(false);
}

void MqttProtocol::PrepareForNetworkStandby() {
    network_standby_preparing_ = true;
    if (reconnect_timer_ != nullptr) {
        esp_timer_stop(reconnect_timer_);
    }
    audio_channel_opened_ = false;
    udp_.reset();
    if (mqtt_ != nullptr) {
        mqtt_->Disconnect();
        mqtt_.reset();
    }
    network_standby_preparing_ = false;
}

bool MqttProtocol::StartMqttClient(bool report_error) {
    if (mqtt_ != nullptr) {
        ESP_LOGW(TAG, "Mqtt client already started");
        audio_channel_opened_ = false;
        udp_.reset();
        mqtt_.reset();
    }

@@ -83,6 +99,9 @@
    mqtt_->SetKeepAlive(keepalive_interval);

    mqtt_->OnDisconnected([this]() {
        if (network_standby_preparing_ || !Board::GetInstance().IsNetworkReady()) {
            return;
        }
        if (on_disconnected_ != nullptr) {
            on_disconnected_();
        }
@@ -143,7 +162,9 @@
    }
    if (!mqtt_->Connect(broker_address, broker_port, client_id, username, password)) {
        ESP_LOGE(TAG, "Failed to connect to endpoint, code=%d", mqtt_->GetLastError());
        SetError(Lang::Strings::SERVER_NOT_CONNECTED);
        if (report_error) {
            SetError(Lang::Strings::SERVER_NOT_CONNECTED);
        }
        return false;
    }

@@ -164,32 +185,60 @@
}

bool MqttProtocol::SendAudio(std::unique_ptr<AudioStreamPacket> packet) {
    std::lock_guard<std::mutex> lock(channel_mutex_);
    if (udp_ == nullptr) {
        return false;
    bool notify_channel_closed = false;

    {
        std::lock_guard<std::mutex> lock(channel_mutex_);
        if (udp_ == nullptr || !audio_channel_opened_) {
            return false;
        }
        if (!Board::GetInstance().IsNetworkReady()) {
            ESP_LOGW(TAG, "Network is not ready, drop UDP audio and close channel");
            audio_channel_opened_ = false;
            udp_.reset();
            notify_channel_closed = true;
        } else {
            std::string nonce(aes_nonce_);
            *(uint16_t*)&nonce[2] = htons(packet->payload.size());
            *(uint32_t*)&nonce[8] = htonl(packet->timestamp);
            *(uint32_t*)&nonce[12] = htonl(++local_sequence_);

            std::string encrypted;
            encrypted.resize(aes_nonce_.size() + packet->payload.size());
            memcpy(encrypted.data(), nonce.data(), nonce.size());

            size_t nc_off = 0;
            uint8_t stream_block[16] = {0};
            if (mbedtls_aes_crypt_ctr(&aes_ctx_, packet->payload.size(), &nc_off, (uint8_t*)nonce.c_str(), stream_block,
                (uint8_t*)packet->payload.data(), (uint8_t*)&encrypted[nonce.size()]) != 0) {
                ESP_LOGE(TAG, "Failed to encrypt audio data");
                return false;
            }

            int ret = udp_->Send(encrypted);
            if (ret <= 0) {
                ESP_LOGW(TAG, "UDP audio send failed, close audio channel");
                audio_channel_opened_ = false;
                udp_.reset();
                notify_channel_closed = true;
            } else {
                return true;
            }
        }
    }

    std::string nonce(aes_nonce_);
    *(uint16_t*)&nonce[2] = htons(packet->payload.size());
    *(uint32_t*)&nonce[8] = htonl(packet->timestamp);
    *(uint32_t*)&nonce[12] = htonl(++local_sequence_);

    std::string encrypted;
    encrypted.resize(aes_nonce_.size() + packet->payload.size());
    memcpy(encrypted.data(), nonce.data(), nonce.size());

    size_t nc_off = 0;
    uint8_t stream_block[16] = {0};
    if (mbedtls_aes_crypt_ctr(&aes_ctx_, packet->payload.size(), &nc_off, (uint8_t*)nonce.c_str(), stream_block,
        (uint8_t*)packet->payload.data(), (uint8_t*)&encrypted[nonce.size()]) != 0) {
        ESP_LOGE(TAG, "Failed to encrypt audio data");
        return false;
    if (notify_channel_closed && on_audio_channel_closed_ != nullptr) {
        on_audio_channel_closed_();
    }

    return udp_->Send(encrypted) > 0;
    return false;
}

void MqttProtocol::CloseAudioChannel(bool send_goodbye) {
    if (!audio_channel_opened_) {
        return;
    }
    audio_channel_opened_ = false;

    {
        std::lock_guard<std::mutex> lock(channel_mutex_);
        udp_.reset();
@@ -287,6 +336,7 @@
    });

    udp_->Connect(udp_server_, udp_port_);
    audio_channel_opened_ = true;

    if (on_audio_channel_opened_ != nullptr) {
        on_audio_channel_opened_();
@@ -385,5 +435,5 @@
}

bool MqttProtocol::IsAudioChannelOpened() const {
    return udp_ != nullptr && !error_occurred_ && !IsTimeout();
    return audio_channel_opened_ && udp_ != nullptr && !error_occurred_ && !IsTimeout();
}