1
// 启动流分发
2
stream_manager_start();

1
// class StreamDispatcher
2
void Start() {
3
    if (running_) return;
4
    running_ = true;
5

6
    // 先启动消费者分发线程（确保它们准备好接收数据）
7
    for (auto& c : consumers_) {
8
        c.thread = std::thread(&StreamDispatcher::DispatchLoop, this, &c);
9
        LOG_DEBUG("Started dispatch thread for consumer: {}", c.name);
10
    }
11

12
    // 短暂延迟，确保消费者线程就绪
13
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
14

15
    // 最后启动获取线程
16
    fetch_thread_ = std::thread(&StreamDispatcher::FetchLoop, this);
17

18
    LOG_INFO("StreamDispatcher started with {} consumers", consumers_.size());
19
}

1
// class StreamDispatcher
2
/**
3
 * @brief 获取循环 - 从 VENC 获取编码流并分发到各队列
4
 *
5
 * 使用零拷贝共享：多个消费者共享同一个 VENC buffer
6
 * shared_ptr 引用计数管理，最后一个使用者释放时调用 ReleaseStream
7
 */
8
void FetchLoop() {
9
    LOG_DEBUG("Fetch loop started for VENC channel {}", venc_chn_id_);
10

11
    uint64_t frameCount = 0;
12
    uint64_t errorCount = 0;
13

14
    while (running_) {
15
        // 从 VENC 获取编码流（零拷贝，shared_ptr 管理生命周期）
16
        auto stream = acquire_encoded_stream(venc_chn_id_, 1000);  // 1秒超时
17

18
        if (!stream) {
19
            errorCount++;
20
            if (errorCount % 5 == 1) {  // 每5次错误打印一次
21
                LOG_WARN("Failed to get stream from VENC channel {} (error #{})",
22
                         venc_chn_id_, errorCount);
23
            }
24
            continue;
25
        }
26

27
        frameCount++;
28
        errorCount = 0;  // 重置错误计数
29

30
        // 打印帧信息
31
        if (frameCount <= 5 || frameCount % 30 == 0) {
32
            LOG_DEBUG("Got frame #{}, size={} bytes",
33
                     frameCount, stream->pstPack ? stream->pstPack->u32Len : 0);
34
        }
35

36
        // 分发给所有消费者（零拷贝，只是增加引用计数）
37
        for (auto& c : consumers_) {
38
            c.queue->push(stream);  // shared_ptr 拷贝，引用计数 +1
39
        }
40

41
        // stream 在这里离开作用域后，引用计数 -1
42
        // 当所有消费者都处理完毕后，最后一个释放时会调用 ReleaseStream
43
    }
44

45
    LOG_DEBUG("Fetch loop exited, total frames: {}", frameCount);
46
}

1
// ============================================================================
2
// 线程安全的媒体队列 - 用于模块间数据分发
3
// ============================================================================
4

5
/**
6
 * @brief 线程安全的媒体缓冲队列
7
 *
8
 * 用于 VENC -> RTSP/WebRTC 的数据分发
9
 * 支持有界队列，超出容量时自动丢弃最旧的帧（背压处理）
10
 *
11
 * @tparam T 媒体数据类型（VideoFramePtr 或 EncodedStreamPtr）
12
 */
13
template <typename T>
14
class MediaQueue {
15
public:
16
      // 具体实现见：https://github.com/Hyrsoft/aipc/blob/main/src/common/media_buffer.h
17
}

1
// class StreamDispatcher
2
/**
3
 * @brief 启动分发器
4
 *
5
 * 启动后会创建：
6
 * 1. 一个 Fetch 线程：从 VENC 获取编码流并直接分发
7
 * 2. Queued 类型消费者的独立处理线程
8
 */
9
void Start() {
10
    if (running_) return;
11
    running_ = true;
12

13
    // 启动 Queued 类型消费者的处理线程
14
    for (auto& c : consumers_) {
15
        if (c.type == ConsumerType::Queued && c.queue) {
16
            c.thread = std::thread(&StreamDispatcher::QueuedConsumerLoop, this, &c);
17
            LOG_DEBUG("Started queued consumer thread for: {}", c.name);
18
        }
19
    }
20

21
    // 短暂延迟，确保 Queued 消费者线程就绪
22
    if (HasQueuedConsumers()) {
23
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
24
    }
25

26
    // 启动 Fetch 线程
27
    fetch_thread_ = std::thread(&StreamDispatcher::FetchLoop, this);
28

29
    LOG_INFO("StreamDispatcher started with {} consumers (optimized for single-core)",
30
             consumers_.size());
31
}

1
/**
2
 * @brief 消费者类型
3
 */
4
enum class ConsumerType {
5
    Direct,     ///< 直接在 Fetch 线程中执行（仅用于极快的操作）
6
    AsyncIO,    ///< 通过 asio::post 投递到 IO 线程执行（网络发送）
7
    Queued      ///< 通过队列投递到独立线程（文件写入等阻塞操作）
8
};
9

10
// class StreamDispatcher
11
struct ConsumerInfo {
12
    std::string name;
13
    StreamConsumer callback;                       // 回调函数
14
    ConsumerType type = ConsumerType::AsyncIO;    // 默认为 AsyncIO 类型
15
    std::unique_ptr<EncodedStreamQueue> queue;    // 仅 Queued 类型使用
16
    std::thread thread;                            // 仅 Queued 类型使用
17
};

1
// class StreamDispatcher
2
/**
3
 * @brief 注册流消费者
4
 *
5
 * @param name 消费者名称（用于日志）
6
 * @param consumer 消费者回调函数
7
 * @param type 消费者类型，决定回调的执行方式
8
 * @param queue_size 队列大小（仅对 Queued 类型有效）
9
 */
10
void RegisterConsumer(const std::string& name, StreamConsumer consumer,
11
                      ConsumerType type = ConsumerType::AsyncIO,
12
                      size_t queue_size = 3) {
13
    ConsumerInfo info;
14
    info.name = name;
15
    info.callback = consumer;
16
    info.type = type;
17

18
    // 只有 Queued 类型需要队列和线程
19
    if (type == ConsumerType::Queued) {
20
        info.queue = std::make_unique<EncodedStreamQueue>(queue_size);
21
    }
22

23
    consumers_.push_back(std::move(info));
24
    LOG_INFO("Registered stream consumer: {} (type={})", name,
25
             type == ConsumerType::Direct ? "Direct" :
26
             type == ConsumerType::AsyncIO ? "AsyncIO" : "Queued");
27
}

1
/**
2
 * @brief 分发帧给所有消费者
3
 *
4
 * 根据消费者类型选择不同的分发策略：
5
 * - Direct: 直接在当前线程调用
6
 * - AsyncIO: 通过 asio::post 投递到 IO 线程
7
 * - Queued: 放入消费者专属队列
8
 */
9
void DispatchToConsumers(EncodedStreamPtr stream) {
10
    for (auto& c : consumers_) {
11
        if (!c.callback) continue;
12

13
        switch (c.type) {
14
            case ConsumerType::Direct:
15
                // 直接在 Fetch 线程中执行（仅用于极快的操作）
16
                c.callback(stream);
17
                break;
18

19
            case ConsumerType::AsyncIO:
20
                // 投递到 IO 线程执行（网络发送）
21
                // 捕获 stream 的拷贝，增加引用计数
22
                PostToIo([callback = c.callback, stream]() {
23
                    callback(stream);
24
                });
25
                break;
26

27
            case ConsumerType::Queued:
28
                // 放入队列，由专属线程处理（文件写入）
29
                if (c.queue) {
30
                    c.queue->push(stream);
31
                }
32
                break;
33
        }
34
    }
35
}

1
// class IoContext
2
/**
3
 * @brief 投递任务到 IO 线程执行
4
 *
5
 * 线程安全，可从任意线程调用
6
 *
7
 * @param handler 要执行的任务
8
 */
9
template<typename Handler>
10
void Post(Handler&& handler) {
11
    asio::post(io_context_, std::forward<Handler>(handler));
12
}

1
case ConsumerType::AsyncIO:
2
    // 将 stream 拷贝一份捕获到 lambda 中
3
    // 并非数据深拷贝，而是 shared_ptr 引用计数自动 +1
4
    PostToIo([callback = c.callback, stream]() {
5
        callback(stream); // 任务执行时，引用计数保证了 buffer 依然有效
6
    });
7
    break;