user/include/app.hpp

#ifndef DATAFLOW_APP_HPP
#define DATAFLOW_APP_HPP

#include <string>
#include <vector>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <stdexcept>

namespace dataflow {

/**
 * @brief 线程安全的队列，用于模块间数据传递。
 *
 * 支持多线程并发入队和出队操作，提供阻塞和非阻塞两种弹出方式。
 */
template<typename T>
class ThreadSafeQueue {
public:
    ThreadSafeQueue() = default;
    ~ThreadSafeQueue() = default;

    ThreadSafeQueue(const ThreadSafeQueue&) = delete;
    ThreadSafeQueue& operator=(const ThreadSafeQueue&) = delete;

    void push(T item) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::move(item));
        cond_.notify_one();
    }

    bool tryPop(T& item) {
        std::lock_guard<std::mutex> lock(mutex_);
        if (queue_.empty()) {
            return false;
        }
        item = std::move(queue_.front());
        queue_.pop();
        return true;
    }

    void waitAndPop(T& item) {
        std::unique_lock<std::mutex> lock(mutex_);
        cond_.wait(lock, [this]() { return !queue_.empty(); });
        item = std::move(queue_.front());
        queue_.pop();
    }

    bool empty() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.empty();
    }

    size_t size() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.size();
    }

    void clear() {
        std::lock_guard<std::mutex> lock(mutex_);
        while (!queue_.empty()) {
            queue_.pop();
        }
    }

private:
    mutable std::mutex mutex_;
    std::queue<T> queue_;
    std::condition_variable cond_;
};

/**
 * @brief 原始数据包结构。
 *
 * 对应自定义 TCP 协议的消息格式：
 * 消息头（4 字节长度字段，小端序）+ 消息体（UTF-8 编码字符串）。
 */
struct RawDataPacket {
    uint32_t length;   ///< 消息体长度（字节数）
    std::string body;  ///< 消息体内容（UTF-8 编码，最大 1024 字符）
};

/**
 * @brief CSU-01：数据接收模块。
 *
 * 负责模拟 TCP 服务端接收外部数据源的数据，
 * 按照自定义协议（消息头 + 消息体）进行拆包解析，
 * 将原始数据字符串传递给数据处理模块。
 */
class DataReceiver {
public:
    /**
     * @brief 构造数据接收模块。
     * @param outputQueue 输出队列，将解析后的原始数据传递给下一模块。
     */
    explicit DataReceiver(ThreadSafeQueue<std::string>& outputQueue);

    ~DataReceiver();

    /**
     * @brief 启动数据接收模块。
     *
     * 启动内部工作线程，从模拟数据源获取数据并解析。
     *
     * @requirement(name="数据接收功能", id="SRS-测试_F-001")
     * 规则：作为 TCP 服务端监听端口，接收数据，按协议拆包。
     */
    void start();

    /**
     * @brief 停止数据接收模块。
     */
    void stop();

    /**
     * @brief 设置模拟数据源。
     * @param data 模拟的外部数据源消息体字符串列表。
     *
     * 每个字符串将被自动封装为符合协议的数据包（4 字节长度前缀 + 消息体）。
     */
    void setMockDataSource(const std::vector<std::string>& data);

    /** @brief 获取已成功接收并解析的数据包数量。 */
    size_t getReceivedCount() const;

    /** @brief 获取解析失败的数据包数量。 */
    size_t getErrorCount() const;

    /** @brief 获取最近一次错误信息。 */
    std::string getLastError() const;

private:
    /**
     * @brief 内部工作线程主循环。
     *
     * 遍历模拟数据源，构造完整数据包并调用 parsePacket 解析，
     * 将解析后的原始数据推入输出队列。
     *
     * @requirement(name="数据接收功能", id="SRS-测试_F-001")
     * 规则：接收网络数据，拆包，传递给数据处理模块。
     */
    void receiveLoop();

    /**
     * @brief 解析数据包。
     * @param buffer  包含完整数据包（消息头 + 消息体）的字节缓冲区。
     * @param offset  当前解析的起始偏移量（传入/传出）。
     * @param outData [out] 解析出的消息体字符串。
     * @return true 解析成功；false 数据不完整或格式错误。
     *
     * @requirement(name="数据接收功能", id="SRS-测试_F-001")
     * 规则：根据约定的协议（消息头+消息体）进行拆包，解决粘包问题。
     */
    bool parsePacket(const std::vector<char>& buffer, size_t& offset, std::string& outData);

    ThreadSafeQueue<std::string>& outputQueue_;
    std::vector<std::string> mockData_;
    std::atomic<bool> running_{false};
    std::thread workerThread_;
    std::atomic<size_t> receivedCount_{0};
    std::atomic<size_t> errorCount_{0};
    std::string lastError_;
    mutable std::mutex errorMutex_;
};

/**
 * @brief CSU-02：数据处理模块。
 *
 * 从内部队列获取原始数据字符串，
 * 进行格式校验（非空、长度限制、字符合法性）和格式转换，
 * 将校验通过的数据传递给数据存储模块。
 */
class DataProcessor {
public:
    /**
     * @brief 构造数据处理模块。
     * @param inputQueue  输入队列，来自 CSU-01 的原始数据。
     * @param outputQueue 输出队列，处理后的数据传递给 CSU-03。
     */
    DataProcessor(ThreadSafeQueue<std::string>& inputQueue,
                  ThreadSafeQueue<std::string>& outputQueue);

    ~DataProcessor();

    /**
     * @brief 启动数据处理模块。
     *
     * 启动内部工作线程，持续从输入队列获取数据进行处理。
     *
     * @requirement(name="数据接收功能", id="SRS-测试_F-001")
     * 规则：对数据进行解析、校验和格式转换。
     */
    void start();

    /**
     * @brief 停止数据处理模块。
     */
    void stop();

    /** @brief 获取成功处理的数据包数量。 */
    size_t getProcessedCount() const;

    /** @brief 因校验失败被丢弃的数据包数量。 */
    size_t getDiscardedCount() const;

    /** @brief 处理过程中发生异常的次数。 */
    size_t getErrorCount() const;

    /** @brief 获取最近一次错误信息。 */
    std::string getLastError() const;

private:
    /**
     * @brief 内部工作线程主循环。
     *
     * 从输入队列获取原始数据，调用 validateAndTransform 进行校验和转换，
     * 通过后推入输出队列，否则丢弃并记录错误。
     *
     * @requirement(name="数据接收功能", id="SRS-测试_F-001")
     * 规则：对数据进行解析、校验（例如完整性校验）和格式转换。
     */
    void processLoop();

    /**
     * @brief 校验并转换原始数据。
     * @param rawData      待校验的原始数据字符串。
     * @param outProcessed [out] 校验通过后的处理结果。
     * @return true 校验通过；false 校验失败。
     *
     * 校验规则：
     * - 字符串非空
     * - 长度不超过 1024 字符
     * - 仅包含可打印 UTF-8 字符（简单 ASCII 可打印范围检查）
     *
     * @requirement(name="数据接收功能", id="SRS-测试_F-001")
     * 规则：校验失败的数据丢弃，记录错误日志。
     */
    bool validateAndTransform(const std::string& rawData, std::string& outProcessed);

    ThreadSafeQueue<std::string>& inputQueue_;
    ThreadSafeQueue<std::string>& outputQueue_;
    std::atomic<bool> running_{false};
    std::thread workerThread_;
    std::atomic<size_t> processedCount_{0};
    std::atomic<size_t> discardedCount_{0};
    std::atomic<size_t> errorCount_{0};
    std::string lastError_;
    mutable std::mutex errorMutex_;
};

/**
 * @brief CSU-03：数据存储模块。
 *
 * 将处理后的数据持久化存储。
 * 当前版本使用内存向量模拟 MySQL 数据库的 t_received_data 表写入。
 */
class DataStorage {
public:
    /**
     * @brief 构造数据存储模块。
     * @param inputQueue 输入队列，来自 CSU-02 的处理后数据。
     */
    explicit DataStorage(ThreadSafeQueue<std::string>& inputQueue);

    ~DataStorage();

    /**
     * @brief 启动数据存储模块。
     *
     * 启动内部工作线程，持续从输入队列获取数据并写入存储。
     *
     * @requirement(name="数据接收功能", id="SRS-测试_F-001")
     * 规则：将处理后的数据持久化写入数据库。
     */
    void start();

    /**
     * @brief 停止数据存储模块。
     */
    void stop();

    /** @brief 成功存储的数据条数。 */
    size_t getStoredCount() const;

    /** @brief 存储过程中发生的错误次数。 */
    size_t getErrorCount() const;

    /** @brief 获取最近一次错误信息。 */
    std::string getLastError() const;

    /**
     * @brief 获取所有已存储的数据。
     * @return 存储的数据字符串列表。
     */
    std::vector<std::string> getAllStoredData() const;

    /** @brief 当前存储的数据总条数。 */
    size_t getStorageSize() const;

private:
    /**
     * @brief 内部工作线程主循环。
     *
     * 从输入队列获取处理后的数据，调用 storeToDatabase 写入存储。
     *
     * @requirement(name="数据接收功能", id="SRS-测试_F-001")
     * 规则：将处理后的数据持久化写入数据库。
     */
    void storageLoop();

    /**
     * @brief 将数据写入存储。
     * @param data 需要存储的数据字符串。
     * @return true 写入成功；false 写入失败。
     *
     * @requirement(name="数据接收功能", id="SRS-测试_F-001")
     * 规则：处理数据库写入失败等异常，并记录日志。
     */
    bool storeToDatabase(const std::string& data);

    ThreadSafeQueue<std::string>& inputQueue_;
    std::atomic<bool> running_{false};
    std::thread workerThread_;
    std::atomic<size_t> storedCount_{0};
    std::atomic<size_t> errorCount_{0};
    std::string lastError_;
    mutable std::mutex errorMutex_;

    /// 内存模拟数据库存储（对应 t_received_data 表的数据内容）
    std::vector<std::string> storage_;
    mutable std::mutex storageMutex_;
};

} // namespace dataflow

#endif // DATAFLOW_APP_HPP