video-pc/AV/code/player/thread_manager.h

#pragma once

#include <thread>
#include <atomic>
#include <memory>
#include <functional>
#include <mutex>
#include "../base/types.h"

extern "C" {
#include <libavformat/avformat.h>
#include <libavcodec/avcodec.h>
#include <libavcodec/bsf.h>
}

namespace av {

// 前向声明codec命名空间中的类
namespace codec {
    class VideoDecoder;
    class AudioDecoder;
}

namespace player {

}

namespace utils {
class Synchronizer;
}

namespace utils {
class PacketQueue;
class FrameQueue;
}

namespace player {

// 使用codec命名空间中的解码器
using VideoDecoder = av::codec::VideoDecoder;
using AudioDecoder = av::codec::AudioDecoder;

// 帧输出回调函数类型定义
using FrameOutputCallback = std::function<void(const AVFramePtr&)>;

/**
 * @brief 线程基类
 */
class ThreadBase {
public:
    ThreadBase(const std::string& name);
    virtual ~ThreadBase();
    
    // 禁用拷贝
    ThreadBase(const ThreadBase&) = delete;
    ThreadBase& operator=(const ThreadBase&) = delete;
    
    /**
     * @brief 启动线程
     */
    bool start();
    
    /**
     * @brief 停止线程
     */
    void stop();
    
    /**
     * @brief 等待线程结束
     */
    void join();
    
    /**
     * @brief 检查线程是否正在运行
     */
    bool isRunning() const;
    
    /**
     * @brief 获取线程名称
     */
    const std::string& getName() const { return m_name; }
    
protected:
    /**
     * @brief 线程主函数，子类需要实现
     */
    virtual void run() = 0;
    
    /**
     * @brief 检查是否应该停止
     */
    bool shouldStop() const { return m_shouldStop.load(); }
    
private:
    std::string m_name;
    std::unique_ptr<std::thread> m_thread;
    std::atomic<bool> m_running{false};
    std::atomic<bool> m_shouldStop{false};
    
    void threadEntry();
};

/**
 * @brief 读取线程，负责从媒体文件读取数据包
 */
class ReadThread : public ThreadBase {
public:
    ReadThread(AVFormatContext* formatContext, 
               av::utils::PacketQueue* packetQueue,
               int videoStreamIndex = -1,
               int audioStreamIndex = -1);
    
    ~ReadThread() override = default;
    
    /**
     * @brief 设置跳转目标
     */
    void seek(int64_t timestamp);
    
protected:
    void run() override;
    
private:
    AVFormatContext* m_formatContext;
    av::utils::PacketQueue* m_packetQueue;
    int m_videoStreamIndex;
    int m_audioStreamIndex;
    std::atomic<int64_t> m_seekTarget{-1};
    std::atomic<bool> m_seeking{false};
    
    bool readPacket();
    void handleSeek();
};

/**
 * @brief 视频解码线程
 */
class VideoDecodeThread : public ThreadBase {
public:
    VideoDecodeThread(av::utils::PacketQueue* packetQueue,
                       av::utils::FrameQueue* frameQueue,
                       VideoDecoder* decoder,
                       av::utils::Synchronizer* synchronizer,
                       int streamIndex,
                       AVCodecParameters* codecParams = nullptr,
                       AVRational timeBase = {1, 1000000});
    
    ~VideoDecodeThread() override;
    
    /**
     * @brief 设置帧输出回调函数
     */
    void setFrameOutputCallback(FrameOutputCallback callback);
    
protected:
    void run() override;
    
private:
    av::utils::PacketQueue* m_packetQueue;
    av::utils::FrameQueue* m_frameQueue;
    VideoDecoder* m_decoder;
    av::utils::Synchronizer* m_synchronizer;
    int m_streamIndex;
    AVBSFContext* m_bsfContext;
    FrameOutputCallback m_frameOutputCallback;
    AVRational m_timeBase;
    
    bool decodeFrame();
    bool initBitStreamFilter(AVCodecParameters* codecParams);
    void cleanupBitStreamFilter();
};

/**
 * @brief 音频解码线程
 */
class AudioDecodeThread : public ThreadBase {
public:
    AudioDecodeThread(av::utils::PacketQueue* packetQueue,
                       av::utils::FrameQueue* frameQueue,
                       AudioDecoder* decoder,
                       av::utils::Synchronizer* synchronizer,
                       int streamIndex,
                       AVRational timeBase = {1, 1000000});
    
    ~AudioDecodeThread() override = default;
    
    /**
     * @brief 设置帧输出回调函数
     */
    void setFrameOutputCallback(FrameOutputCallback callback);
    
protected:
    void run() override;
    
private:
    av::utils::PacketQueue* m_packetQueue;
    av::utils::FrameQueue* m_frameQueue;
    AudioDecoder* m_decoder;
    av::utils::Synchronizer* m_synchronizer;
    int m_streamIndex;
    FrameOutputCallback m_frameOutputCallback;
    AVRational m_timeBase;
    
    bool decodeFrame();
};

/**
 * @brief 线程管理器，统一管理所有播放相关线程
 */
class ThreadManager {
public:
    ThreadManager();
    ~ThreadManager();
    
    // 禁用拷贝
    ThreadManager(const ThreadManager&) = delete;
    ThreadManager& operator=(const ThreadManager&) = delete;
    
    /**
     * @brief 创建读取线程
     */
    bool createReadThread(AVFormatContext* formatContext,
                          av::utils::PacketQueue* packetQueue,
                          int videoStreamIndex = -1,
                          int audioStreamIndex = -1);
    
    /**
     * @brief 创建视频解码线程
     */
    bool createVideoDecodeThread(av::utils::PacketQueue* packetQueue,
                                  av::utils::FrameQueue* frameQueue,
                                  VideoDecoder* decoder,
                                  av::utils::Synchronizer* synchronizer,
                                  int streamIndex,
                                  AVCodecParameters* codecParams = nullptr,
                                  AVRational timeBase = {1, 1000000});
    
    /**
     * @brief 创建音频解码线程
     */
    bool createAudioDecodeThread(av::utils::PacketQueue* packetQueue,
                                  av::utils::FrameQueue* frameQueue,
                                  AudioDecoder* decoder,
                                  av::utils::Synchronizer* synchronizer,
                                  int streamIndex,
                                  AVRational timeBase = {1, 1000000});
    
    /**
     * @brief 启动所有线程
     */
    bool startAll();
    
    /**
     * @brief 停止所有线程
     */
    void stopAll();
    
    /**
     * @brief 等待所有线程结束
     */
    void joinAll();
    
    /**
     * @brief 检查是否有线程正在运行
     */
    bool hasRunningThreads() const;
    
    /**
     * @brief 获取读取线程
     */
    ReadThread* getReadThread() const { return m_readThread.get(); }
    
    /**
     * @brief 获取视频解码线程
     */
    VideoDecodeThread* getVideoDecodeThread() const { return m_videoDecodeThread.get(); }
    
    /**
     * @brief 获取音频解码线程
     */
    AudioDecodeThread* getAudioDecodeThread() const { return m_audioDecodeThread.get(); }
    
private:
    std::unique_ptr<ReadThread> m_readThread;
    std::unique_ptr<VideoDecodeThread> m_videoDecodeThread;
    std::unique_ptr<AudioDecodeThread> m_audioDecodeThread;
};

/**
 * @brief 线程安全的回调函数包装器
 */
template<typename Func>
class ThreadSafeCallback {
public:
    ThreadSafeCallback(Func&& func) : m_func(std::forward<Func>(func)) {}
    
    template<typename... Args>
    auto operator()(Args&&... args) -> decltype(m_func(std::forward<Args>(args)...)) {
        std::lock_guard<std::mutex> lock(m_mutex);
        return m_func(std::forward<Args>(args)...);
    }
    
private:
    Func m_func;
    mutable std::mutex m_mutex;
};

/**
     * @brief 创建线程安全的回调函数
     */
template<typename Func>
auto makeThreadSafeCallback(Func&& func) {
    return ThreadSafeCallback<Func>(std::forward<Func>(func));
}

} // namespace player
} // namespace av