update ffmpeg_speaker

src/devices/speaker/ffmpeg_speaker/CMakeLists.txt

@@ -6,4 +6,4 @@ find_package(ALSA REQUIRED)
 
 add_library(cmvr_es::device::ffmpeg_speaker ALIAS ffmpeg_speaker)
 
-target_link_libraries(ffmpeg_speaker PRIVATE ALSA::ALSA)
+target_link_libraries(ffmpeg_speaker PRIVATE -lpulse-simple -lpulse)

src/devices/speaker/ffmpeg_speaker/include/ffmpeg_speaker.h

@@ -6,10 +6,13 @@
 #include <thread>
 #include <mutex>
 #include <condition_variable>
-#include "../../abstract_speaker.h"
+#include "speaker/abstract_speaker.h"
 #include <boost/lockfree/spsc_queue.hpp>
+#include <pulse/simple.h>
+#include <pulse/error.h>
+#include <pulse/pulseaudio.h>
 #include "ffmpeg_ptr.h"
-#include "../../../../utils/base/include/os.h"
+#include "utils/base/include/os.h"
 
 extern "C" {
     #include <alsa/asoundlib.h>
@@ -37,24 +40,36 @@ namespace cmvr::device {
         void getState(SpeakerState& state) override;
 
         void resetPlayState();
+        bool initPulseDevice_();
+        bool initAudioParams_(const std::string& audio_path);
     private:
         void decode_audio_();
         void play_audio_();
 
         std::string id_;
-        std::string alas_;
+        pa_simple* pulse_simple_ = nullptr;  // 修改：PulseAudio 简单 API 句柄
-        int channels_;
+        pa_sample_spec sample_spec_{};         // 新增：PulseAudio 采样规格
-        int sample_rate_;
+        int channels_ = 0;
-        int soft_resample_;
+        int sample_rate_ = 0;
-        int latency_;
+        int soft_resample_ = 0;
-        snd_pcm_t* pcm_handle_ = nullptr;
+        int latency_ = 0;
 
         std::shared_ptr<std::thread> decode_thread_;
         std::shared_ptr<std::thread> play_thread_;
         std::mutex mtx_;
         std::string audio_path_;
 
+        // Boost 单生产者单消费者无锁队列
         boost::lockfree::spsc_queue<AudioFrame, boost::lockfree::capacity<AUDIO_QUEUE_CAPACITY>> audio_queue_;
+
+        // 暂停相关的同步变量
+        mutable std::mutex mtx_pause_;
+        std::condition_variable cv_pause_;
+        std::atomic<bool> is_paused_{false};
+
+        // 音频时钟同步
+        std::atomic<int64_t> audio_pts_{0};
+        std::chrono::time_point<std::chrono::steady_clock> playback_start_time_;
     };
 
 }

src/devices/speaker/ffmpeg_speaker/src/ffmpeg_speaker.cpp

@@ -13,11 +13,12 @@ ffmpegSpeaker::ffmpegSpeaker(const XmlNode& cfg) : AbstractSpeaker(cfg) {
     state_.is_paused = false;
     try {
         id_ = cfg_.getAttrString("id");
-        alas_ = cfg_.getAttrString("alas");
+        memset(&sample_spec_, 0, sizeof(sample_spec_));
-        channels_ = std::clamp(cfg_.getAttrDefault("channels", 2), 1, 8);
+        // alas_ = cfg_.getAttrString("alas");
-        sample_rate_ = cfg_.getAttrDefault("sampleRate", 44100);
+        // channels_ = std::clamp(cfg_.getAttrDefault("channels", 2), 1, 8);
-        soft_resample_ = cfg_.getAttrDefault("softResample", 1);
+        // sample_rate_ = cfg_.getAttrDefault("sampleRate", 44100);
-        latency_ = cfg_.getAttrDefault("latency", 500000);
+        // soft_resample_ = cfg_.getAttrDefault("softResample", 1);
+        // latency_ = cfg_.getAttrDefault("latency", 500000);
         state_.volume = clamp(cfg_.getAttrDefault("volume", 100), 0, 100);
     }
     catch (const exception& e) {
@@ -26,7 +27,27 @@ ffmpegSpeaker::ffmpegSpeaker(const XmlNode& cfg) : AbstractSpeaker(cfg) {
 }
 
 ffmpegSpeaker::~ffmpegSpeaker() {
-    stop();
+    {
+        std::lock_guard<std::mutex> lock(mtx_);
+        state_.is_running = false;
+        state_.is_decoding = false;
+        state_.is_paused = false;
+    }
+
+    cv_pause_.notify_all(); // 唤醒所有等待的线程
+
+    // 等待线程结束
+    if (decode_thread_ && decode_thread_->joinable()) {
+        decode_thread_->join();
+        decode_thread_.reset();
+    }
+
+    if (play_thread_ && play_thread_->joinable()) {
+        play_thread_->join();
+        play_thread_.reset();
+    }
+
+    resetPlayState();
 }
 
 void ffmpegSpeaker::init() {
@@ -47,84 +68,46 @@ void ffmpegSpeaker::init() {
 }
 
 void ffmpegSpeaker::start() {
-    try {
+
-        lock_guard lock(mtx_);
-        if (!state_.is_initialized) {
-            LOG(ERROR) << "[ffmpegSpeaker] (start): device not initialized, id=" + id_;
-            throw runtime_error("[ffmpegSpeaker] (start): device not initialized, id=" + id_);
-        }
-        if (state_.is_running) {
-            LOG(WARNING) << "[ffmpegSpeaker] (start): device already started, id=" + id_;
-            throw runtime_error("[ffmpegSpeaker] (start): device already started, id=" + id_);
-        }
-        if (snd_pcm_open(&pcm_handle_, alas_.c_str(), SND_PCM_STREAM_PLAYBACK, 0) < 0) {
-            LOG(ERROR) << "[ffmpegSpeaker] (init): Failed to open ALSA device: " << alas_;
-            throw runtime_error("[ffmpegSpeaker] (init): Failed to open ALSA device: " + alas_);
-        }
-        if (snd_pcm_set_params(pcm_handle_, SND_PCM_FORMAT_S16_LE, SND_PCM_ACCESS_RW_INTERLEAVED,
-            channels_, sample_rate_, soft_resample_, latency_) < 0) {
-            LOG(ERROR) << "[ffmpegSpeaker] (init): Failed to set ALSA params, id=" << id_;
-            throw runtime_error("[ffmpegSpeaker] (init): Failed to set ALSA params, id=" + id_);
-        }
-        state_.is_running = false;
-        state_.is_paused = false;
-        state_.is_decoding = false;
-        LOG(INFO) << "[ffmpegSpeaker] (start): Success, id=" << id_;
-    }
-    catch (const exception& e) {
-        LOG(ERROR) <<"[ffmpegSpeaker] (start): Failed, id=" << id_ << ": " << e.what();
-        throw runtime_error("[ffmpegSpeaker] (start): Failed, id=" + id_ + ": " + string(e.what()));
-    }
 }
+
 void ffmpegSpeaker::resetPlayState()
 {
-    if (pcm_handle_) {
+    if (pulse_simple_) {
-        snd_pcm_drain(pcm_handle_);
+        pa_simple_flush(pulse_simple_, nullptr);
-        snd_pcm_close(pcm_handle_);
+        pa_simple_free(pulse_simple_);
-        pcm_handle_ = nullptr;
+        pulse_simple_ = nullptr;
     }
-    state_.is_decoding = false;
+
-    state_.is_running = false;
+    // 清空队列
-    state_.is_paused = false;
+    AudioFrame frame;
+    while (audio_queue_.pop(frame)) {
+        // 清空所有帧
+    }
+
+    state_.is_initialized = false;
+
     audio_path_.clear();
-
-
     LOG(INFO) << "[ffmpegSpeaker] (resetPlayState): Success, id=" << id_;
 }
 
 void ffmpegSpeaker::stop() {
-    try {
+
-        //此处要注意可能和解码线程冲突引起死锁
+    state_.is_running = false;
-        {
+    state_.is_decoding = false;
-            lock_guard lock(mtx_);
+    state_.is_paused = false;
-            if (!state_.is_running) {
+    // 等待线程结束
-                LOG(WARNING) << "[ffmpegSpeaker] (start): device already stoped, id=" + id_;
+    if (decode_thread_ && decode_thread_->joinable()) {
-                throw runtime_error("[ffmpegSpeaker] (start): device already stoped, id=" + id_);
+        decode_thread_->join();
-            }
+        decode_thread_.reset();
-            state_.is_decoding = false;
-            state_.is_running = false;
-            state_.is_paused = false;
-            audio_path_.clear();
-        }
-        if (decode_thread_->joinable()) {
-            decode_thread_->join();
-            decode_thread_.reset();
-        }
-        if (play_thread_->joinable()) {
-            play_thread_->join();
-            play_thread_.reset();
-        }
-        if (pcm_handle_) {
-            snd_pcm_drain(pcm_handle_);
-            snd_pcm_close(pcm_handle_);
-            pcm_handle_ = nullptr;
-        }
-        LOG(INFO) << "[ffmpegSpeaker] (stop): Success, id=" << id_;
     }
-    catch (const exception& e) {
+
-        LOG(ERROR) <<"[ffmpegSpeaker] (stop): Failed, id=" << id_ << ": " << e.what();
+    if (play_thread_ && play_thread_->joinable()) {
-        throw runtime_error("[ffmpegSpeaker] (stop): Failed, id=" + id_ + ": " + string(e.what()));
+        play_thread_->join();
+        play_thread_.reset();
     }
+
+    resetPlayState();
 }
 
 void ffmpegSpeaker::pause() {
@@ -169,42 +152,91 @@ void ffmpegSpeaker::resume() {
     }
 }
 
-void ffmpegSpeaker::play(const string& audio_path) {
+void ffmpegSpeaker::play(const std::string& audio_path) {
+    std::unique_lock<std::mutex> lock(mtx_);
+
     try {
-        lock_guard lock(mtx_);
         if (state_.is_running) {
             LOG(WARNING) << "[ffmpegSpeaker] (play): Speaker is playing " << audio_path_ << ", id=" << id_;
             throw runtime_error("[ffmpegSpeaker] (play): Speaker is playing " + audio_path_);
         }
         if (!pathExists(audio_path)) {
-            snd_pcm_drain(pcm_handle_);
+            throw std::runtime_error("[ffmpegSpeaker] (play): Audio file not found: " + audio_path);
-            snd_pcm_close(pcm_handle_);
+        }
-            LOG(ERROR) << "[ffmpegSpeaker] (play): audio path not exists, id=" << id_ << ", path=" << audio_path;
+        // 如果正在播放，先停止
-            throw runtime_error("[ffmpegSpeaker] (play): audio path not exists, id=" + id_ + ", path=" + audio_path);
+        state_.is_running = false;
+        lock.unlock();
+
+        cv_pause_.notify_all();
+
+        if (decode_thread_ && decode_thread_->joinable()) {
+            decode_thread_->join();
+            decode_thread_.reset();
+        }
+
+        if (play_thread_ && play_thread_->joinable()) {
+            play_thread_->join();
+            play_thread_.reset();
+        }
+
+        resetPlayState();
+        lock.lock();
+        // 清空队列
+        AudioFrame frame;
+        while (audio_queue_.pop(frame)) {
+            // 清空所有帧
         }
 
         audio_path_ = audio_path;
+
+        // 初始化音频参数
+        if (!initAudioParams_(audio_path)) {
+            throw std::runtime_error("[ffmpegSpeaker] Failed to get audio parameters");
+        }
+
+        // 初始化 PulseAudio 设备
+        if (!initPulseDevice_()) {
+            throw std::runtime_error("[ffmpegSpeaker] PulseAudio device initialization failed");
+        }
+
+        // 设置状态
         state_.is_running = true;
         state_.is_decoding = true;
-        if (decode_thread_.get()) {
+        state_.is_paused = false;
-            if (decode_thread_->joinable()) {
+
-                decode_thread_->join();
+        // 启动解码线程
-                decode_thread_.reset();
+        decode_thread_ = std::make_shared<std::thread>([this]() {
+            try {
+                this->decode_audio_();
+            } catch (const std::exception& e) {
+                std::cerr << "[ffmpegSpeaker] Decode thread exception: " << e.what() << std::endl;
+                std::lock_guard<std::mutex> lock(mtx_);
+                state_.is_decoding = false;
             }
-        }
+        });
-        if (play_thread_.get()) {
+
-            if (play_thread_->joinable()) {
+        // 等待解码线程开始（确保有数据）
-                play_thread_->join();
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
-                play_thread_.reset();
+
+        // 启动播放线程
+        play_thread_ = std::make_shared<std::thread>([this]() {
+            try {
+                this->play_audio_();
+            } catch (const std::exception& e) {
+                std::cerr << "[ffmpegSpeaker] Play thread exception: " << e.what() << std::endl;
+                std::lock_guard<std::mutex> lock(mtx_);
+                state_.is_running = false;
             }
-        }
+        });
-        decode_thread_ = make_shared<thread>(&ffmpegSpeaker::decode_audio_, this);
+
-        play_thread_ = make_shared<thread>(&ffmpegSpeaker::play_audio_, this);
+        std::cout << "[ffmpegSpeaker] Started playing: " << audio_path << std::endl;
-        LOG(INFO) << "[ffmpegSpeaker] (play): Success, id=" << id_;
     }
-    catch (const exception& e) {
+    catch (const std::exception& e) {
-        LOG(ERROR) <<"[ffmpegSpeaker] (play): Failed, id=" << id_ << ": " << e.what();
+        state_.is_running = false;
-        throw runtime_error("[ffmpegSpeaker] (play): Failed, id=" + id_ + ": " + string(e.what()));
+        state_.is_decoding = false;
+        lock.unlock();
+        resetPlayState();
+        throw;
     }
 }
 
@@ -216,6 +248,7 @@ void ffmpegSpeaker::setVolume(const int volume) {
 int ffmpegSpeaker::getVolume() const {
     return state_.volume;
 }
+
 void ffmpegSpeaker::getState(SpeakerState& state) {
     state.is_decoding = state_.is_decoding;
     state.volume = state_.volume;
@@ -227,139 +260,267 @@ void ffmpegSpeaker::getState(SpeakerState& state) {
 
 void ffmpegSpeaker::decode_audio_() {
     using namespace ffmpeg;
+
     if (audio_path_.empty()) {
-        LOG(ERROR) << "[ffmpegSpeaker] (decode_audio_): audio_path is empty";
+        {
-        { lock_guard lock(mtx_); state_.is_decoding = false; }
+            std::lock_guard<std::mutex> lock(mtx_);
-        throw runtime_error("[ffmpegSpeaker] (decode_audio_): audio_path is empty");
+            state_.is_decoding = false;
+        }
+        std::cerr << "[ffmpegSpeaker] (decode_audio_): audio_path is empty" << std::endl;
+        return;
     }
 
+    std::cout << "[ffmpegSpeaker] Decoding with: rate=" << sample_rate_
+              << ", channels=" << channels_ << std::endl;
+
     const AVFormatContextPtr fmt_ctx = make_format_context(audio_path_);
     if (!fmt_ctx) {
-        LOG(ERROR) << "[ffmpegSpeaker] (decode_audio_): Failed to open format context";
+        {
-        { lock_guard lock(mtx_); state_.is_decoding = false; }
+            std::lock_guard<std::mutex> lock(mtx_);
-        throw runtime_error("[ffmpegSpeaker] (decode_audio_): Failed to open format context");
+            state_.is_decoding = false;
+        }
+        std::cerr << "[ffmpegSpeaker] (decode_audio_): Failed to open format context" << std::endl;
+        return;
     }
 
     AVCodec* decoder = nullptr;
     const int stream_index = av_find_best_stream(fmt_ctx.get(), AVMEDIA_TYPE_AUDIO, -1, -1, &decoder, 0);
     if (stream_index < 0) {
-        LOG(ERROR) << "[ffmpegSpeaker] (decode_audio_): No audio stream found in file";
+        {
-        { lock_guard lock(mtx_); state_.is_decoding = false; }
+            std::lock_guard<std::mutex> lock(mtx_);
-        throw runtime_error("[ffmpegSpeaker] (decode_audio_): No audio stream found in file");
+            state_.is_decoding = false;
+        }
+        std::cerr << "[ffmpegSpeaker] (decode_audio_): No audio stream found in file" << std::endl;
+        return;
     }
 
     AVStream* audio_stream = fmt_ctx->streams[stream_index];
     const AVCodecContextPtr codec_ctx = make_codec_context(audio_stream, decoder);
     if (!codec_ctx) {
-        LOG(ERROR) << "[ffmpegSpeaker] (decode_audio_): Failed to create codec context";
+        {
-        { lock_guard lock(mtx_); state_.is_decoding = false; }
+            std::lock_guard<std::mutex> lock(mtx_);
-        throw runtime_error("[ffmpegSpeaker] (decode_audio_): Failed to create codec context");
+            state_.is_decoding = false;
+        }
+        std::cerr << "[ffmpegSpeaker] (decode_audio_): Failed to create codec context" << std::endl;
+        return;
     }
 
     const SwrContextPtr swr_ctx = make_swr_context(codec_ctx.get(), channels_, sample_rate_);
     if (!swr_ctx) {
-        LOG(ERROR) << "[ffmpegSpeaker] (decode_audio_): Failed to init swr context";
+        {
-        { lock_guard lock(mtx_); state_.is_decoding = false; }
+            std::lock_guard<std::mutex> lock(mtx_);
-        throw runtime_error("[ffmpegSpeaker] (decode_audio_): Failed to init swr context");
+            state_.is_decoding = false;
+        }
+        std::cerr << "[ffmpegSpeaker] (decode_audio_): Failed to init swr context" << std::endl;
+        return;
     }
 
     const AVPacketPtr pkt(av_packet_alloc());
     const AVFramePtr frame(av_frame_alloc());
 
-    LOG(INFO) << "[ffmpegSpeaker] (decode_audio_): Start decoding " << audio_path_;
+    // 解码循环
-    int64_t start_time = av_gettime(); // 播放开始时间
+    while (true) {
-    while (state_.is_running && state_.is_decoding) {
+        {
+            std::lock_guard<std::mutex> lock(mtx_);
+            if (!state_.is_running || !state_.is_decoding) {
+                break;
+            }
+        }
 
-        if (av_read_frame(fmt_ctx.get(), pkt.get()) < 0) break;
+        // 暂停处理
+        {
+            std::unique_lock<std::mutex> lock(mtx_pause_);
+            cv_pause_.wait(lock, [this]() {
+                return !state_.is_paused || !state_.is_running;
+            });
+            if (!state_.is_running) break;
+        }
+
+        // 控制队列大小，避免内存占用过高
+        if (audio_queue_.write_available() < 5) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(10));
+            continue;
+        }
+
+        if (av_read_frame(fmt_ctx.get(), pkt.get()) < 0) {
+            // 文件读取结束
+            break;
+        }
 
         if (pkt->stream_index == stream_index) {
-            if (avcodec_send_packet(codec_ctx.get(), pkt.get()) < 0) continue;
+            if (avcodec_send_packet(codec_ctx.get(), pkt.get()) < 0) {
+                av_packet_unref(pkt.get());
+                continue;
+            }
 
-            while (avcodec_receive_frame(codec_ctx.get(), frame.get()) == 0) {
+            while (true) {
-                while (state_.is_paused)
                 {
-                    av_usleep(10);
+                    std::lock_guard<std::mutex> lock(mtx_);
-                }
+                    if (!state_.is_running || !state_.is_decoding) {
-                // 计算帧的播放时间(微秒)
+                        av_packet_unref(pkt.get());
-                int64_t frame_time = av_rescale_q(frame->pts,
+                        goto end_decode;
-                                                 fmt_ctx->streams[stream_index]->time_base,
+                    }
-                                                 AV_TIME_BASE_Q);
-                // 计算当前播放进度
-                int64_t current_time = av_gettime() - start_time;
-
-                // 如果解码太快，等待
-                if (frame_time > current_time) {
-                    int64_t sleep_time = frame_time - current_time;
-                    //LOG(INFO) << "准备休眠: " << sleep_time << "us, is_running=" << state_.is_running;
-                    av_usleep(sleep_time);
-                    //LOG(INFO) << "休眠结束, is_running=" << state_.is_running;
                 }
 
+                int ret = avcodec_receive_frame(codec_ctx.get(), frame.get());
+                if (ret == AVERROR(EAGAIN) || ret == AVERROR_EOF) {
+                    break;
+                } else if (ret < 0) {
+                    break;
+                }
+
+                // 计算输出样本数
                 int64_t max_samples64 = av_rescale_rnd(
                     swr_get_delay(swr_ctx.get(), codec_ctx->sample_rate) + frame->nb_samples,
                     sample_rate_, codec_ctx->sample_rate, AV_ROUND_UP);
 
                 const int max_samples = static_cast<int>(std::min<int64_t>(max_samples64, INT_MAX));
 
-                auto buffer = make_shared<vector<int16_t>>(max_samples * 2); // 2 channels
+                // 创建音频帧
+                auto buffer = std::make_shared<std::vector<int16_t>>(max_samples * channels_);
                 uint8_t* out[] = { reinterpret_cast<uint8_t*>(buffer->data()), nullptr };
 
+                // 重采样
                 const int out_samples = swr_convert(
                     swr_ctx.get(),
                     out, max_samples,
                     const_cast<const uint8_t**>(frame->data), frame->nb_samples);
 
-                int16_t* pcm_data = buffer->data();
+                if (out_samples <= 0) {
+                    continue;
+                }
+
+                // 调整大小到实际样本数
+                buffer->resize(out_samples * channels_);
+
+                // 应用音量
                 float volume_scale;
-                { lock_guard lock(mtx_); volume_scale = static_cast<float>(state_.volume) / 100.0f;}
+                {
-                const int sample_count = out_samples * 2;
+                    std::lock_guard<std::mutex> lock(mtx_);
+                    volume_scale = static_cast<float>(state_.volume) / 100.0f;
+                }
+
+                int16_t* pcm_data = buffer->data();
+                const int sample_count = buffer->size();
 
                 for (int i = 0; i < sample_count; ++i) {
                     float scaled = static_cast<float>(pcm_data[i]) * volume_scale;
-                    pcm_data[i] = static_cast<int16_t>(clamp(scaled, -32768.f, 32767.f));
+                    pcm_data[i] = static_cast<int16_t>(std::clamp(scaled, -32768.f, 32767.f));
                 }
 
-                if (!audio_queue_.push(buffer)) {
+                // 推入队列（使用非阻塞方式）
-                    LOG(WARNING) << "[ffmpegSpeaker] (decode_audio_): Audio queue full, drop frame";
+                bool pushed = false;
+                int attempts = 0;
+                while (!pushed && attempts < 10) {
+                    pushed = audio_queue_.push(buffer);
+                    if (!pushed) {
+                        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+                        attempts++;
+                    }
+                }
+
+                if (!pushed) {
+                    std::cerr << "[ffmpegSpeaker] Failed to push audio frame to queue after 10 attempts" << std::endl;
                 }
             }
         }
         av_packet_unref(pkt.get());
     }
-    LOG(INFO) << "[ffmpegSpeaker] (decode_audio_): End of decoding";
+
-    state_.is_decoding = false;
+end_decode:
+    {
+        std::lock_guard<std::mutex> lock(mtx_);
+        state_.is_decoding = false;
+    }
+    std::cout << "[ffmpegSpeaker] Decoding finished" << std::endl;
 }
 
 void ffmpegSpeaker::play_audio_() {
-    while (state_.is_running) {
+    int pa_error = 0;
-        if (state_.is_paused) {
+
-            this_thread::sleep_for(chrono::milliseconds(500));
+    while (true) {
-            continue;
+        {
-        }
+            std::lock_guard<std::mutex> lock(mtx_);
-        AudioFrame frame;
+            if (!state_.is_running) {
-        if (audio_queue_.pop(frame)) {
+                break;
-            const auto samples = frame->size() / 2; // stereo
-            if (snd_pcm_writei(pcm_handle_, frame->data(), samples) < 0) {
-                snd_pcm_prepare(pcm_handle_); // 恢复播放状态
-                LOG(WARNING) << "[ffmpegSpeaker] ALSA buffer underrun, recover";
             }
         }
-        else {
+
-            //音频数据队列为空且解码已经结束，退出
+        // 暂停处理
-            if (!state_.is_decoding) {
+        if (state_.is_paused) {
-                if (pcm_handle_) {
+            std::unique_lock<std::mutex> lock(mtx_pause_);
-                    // 停止播放,清空缓存，否则结束会有杂音
+            cv_pause_.wait(lock, [this]() {
-                    snd_pcm_drop(pcm_handle_);
+                return !state_.is_paused || !state_.is_running;
-                    LOG(INFO) << "[ffmpegSpeaker] ALSA buffer cleared";
+            });
+            if (!state_.is_running) break;
+        }
+
+        AudioFrame frame;
+
+        // 尝试获取音频帧
+        if (!audio_queue_.pop(frame)) {
+            // 检查解码是否结束
+            bool decoding_finished = false;
+            {
+                std::lock_guard<std::mutex> lock(mtx_);
+                decoding_finished = !state_.is_decoding;
+            }
+
+            if (decoding_finished && audio_queue_.empty()) {
+                break;  // 解码结束且队列为空，播放完成
+            }
+
+            // 短暂等待后重试
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            continue;
+        }
+
+        // 使用 PulseAudio 播放
+        if (pulse_simple_ && !frame->empty()) {
+            const size_t data_size = frame->size() * sizeof(int16_t);
+            int ret = pa_simple_write(pulse_simple_, frame->data(), data_size, &pa_error);
+
+            if (ret < 0) {
+                std::cerr << "[ffmpegSpeaker] (play_audio_): PulseAudio write failed: "
+                         << pa_strerror(pa_error) << std::endl;
+
+                // 尝试重新连接
+                resetPlayState();
+
+                // 稍等后重试初始化
+                std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+                if (!initPulseDevice_()) {
+                    std::cerr << "[ffmpegSpeaker] (play_audio_): Failed to reconnect" << std::endl;
+                    {
+                        std::lock_guard<std::mutex> lock(mtx_);
+                        state_.is_running = false;
+                    }
+                    break;
+                }
+
+                // 重新写入当前帧
+                ret = pa_simple_write(pulse_simple_, frame->data(), data_size, &pa_error);
+                if (ret < 0) {
+                    std::lock_guard<std::mutex> lock(mtx_);
+                    state_.is_running = false;
                     break;
                 }
             }
-            this_thread::sleep_for(chrono::seconds(1)); // 空队列等待
         }
     }
-    LOG(INFO) << "[ffmpegSpeaker] (play_audio_): Playback thread exiting";
+
-    //state_.is_running = false;
+    // 播放结束，刷新缓冲区
-    resetPlayState();
+    if (pulse_simple_) {
+        pa_simple_flush(pulse_simple_, &pa_error);
+    }
+
+    {
+        std::lock_guard<std::mutex> lock(mtx_);
+        state_.is_running = false;
+    }
+
+    std::cout << "[ffmpegSpeaker] Playback finished" << std::endl;
 }
 
 void ffmpegSpeaker::updateParams(const std::pair<std::string, std::string>& param)
@@ -381,3 +542,87 @@ void ffmpegSpeaker::updateParams(const std::pair<std::string, std::string>& para
         throw runtime_error(e.what());
     }
 }
+
+bool ffmpegSpeaker::initPulseDevice_() {
+    // 验证参数
+    if (sample_rate_ <= 0 || channels_ <= 0) {
+        std::cerr << "[ffmpegSpeaker] (initPulseDevice_): Invalid audio parameters: rate="
+                  << sample_rate_ << ", channels=" << channels_ << std::endl;
+        return false;
+    }
+
+    // 配置采样规格
+    sample_spec_.format = PA_SAMPLE_S16LE;      // 16位有符号小端整数
+    sample_spec_.rate = sample_rate_;           // 采样率
+    sample_spec_.channels = channels_;          // 声道数
+
+    // 检查采样规格是否有效
+    if (!pa_sample_spec_valid(&sample_spec_)) {
+        std::cerr << "[ffmpegSpeaker] (initPulseDevice_): Invalid sample specification: "
+                  << "rate=" << sample_spec_.rate
+                  << ", channels=" << sample_spec_.channels << std::endl;
+        return false;
+    }
+
+    // 释放现有的连接
+    if (pulse_simple_) {
+        pa_simple_free(pulse_simple_);
+        pulse_simple_ = nullptr;
+    }
+
+    // 创建 PulseAudio 简单连接
+    int pa_error = 0;
+    pulse_simple_ = pa_simple_new(
+        nullptr,                    // 使用默认服务器
+        "ffmpegSpeaker",           // 应用程序名称
+        PA_STREAM_PLAYBACK,        // 播放流
+        nullptr,                    // 使用默认设备
+        "Music",                   // 流描述
+        &sample_spec_,             // 采样规格
+        nullptr,                    // 使用默认通道映射
+        nullptr,                    // 缓冲区属性（使用默认）
+        &pa_error                  // 错误码
+    );
+
+    if (!pulse_simple_) {
+        std::cerr << "[ffmpegSpeaker] (initPulseDevice_): Failed to create PulseAudio connection: "
+                 << pa_strerror(pa_error) << std::endl;
+        return false;
+    }
+
+    state_.is_initialized = true;
+    std::cout << "[ffmpegSpeaker] PulseAudio initialized: rate=" << sample_rate_
+              << ", channels=" << channels_ << std::endl;
+    return true;
+}
+
+bool ffmpegSpeaker::initAudioParams_(const std::string& audio_path) {
+    using namespace ffmpeg;
+
+    auto fmt_ctx = make_format_context(audio_path);
+    if (!fmt_ctx) {
+        std::cerr << "[ffmpegSpeaker] Failed to open audio file: " << audio_path << std::endl;
+        return false;
+    }
+
+    int stream_idx = av_find_best_stream(fmt_ctx.get(), AVMEDIA_TYPE_AUDIO, -1, -1, nullptr, 0);
+    if (stream_idx < 0) {
+        std::cerr << "[ffmpegSpeaker] No audio stream found" << std::endl;
+        return false;
+    }
+
+    auto codec_par = fmt_ctx->streams[stream_idx]->codecpar;
+    channels_ = codec_par->channels;
+    sample_rate_ = codec_par->sample_rate;
+
+    // 确保合理的参数
+    if (channels_ <= 0) channels_ = 2;
+    if (sample_rate_ <= 0) sample_rate_ = 44100;
+
+    // 检查并打印详细的音频参数
+    std::cout << "[ffmpegSpeaker] Audio parameters: rate=" << sample_rate_
+              << ", channels=" << channels_
+              << ", format=" << av_get_sample_fmt_name((AVSampleFormat)codec_par->format)
+              << std::endl;
+    return true;
+}

src/service/src/grpc_speaker_service.cpp

@@ -41,7 +41,7 @@ grpc::Status gRPCSpeakerServiceImpl::PlayAudio(grpc::ServerContext* context,
         string dev_id = request->header().device_id();
         LOG(INFO) << "[gRPCSpeakerServiceImpl] (PlayAudio): id=" << dev_id;
         const auto dev = dmgr_.getDevice<AbstractSpeaker>(dev_id);
-        dev->start();
+        //dev->start();
         dev->play(request->audio_path());
         response->mutable_header()->set_success(true);
         setCurrentTimestamp(response->mutable_header()->mutable_timestamp());