生成代码工程

CMakeLists.txt

@@ -0,0 +1,37 @@
+cmake_minimum_required(VERSION 3.14)
+project(dataflow_system VERSION 1.0.0 LANGUAGES CXX)
+
+# C++17 标准
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+set(CMAKE_CXX_EXTENSIONS OFF)
+
+# MSVC UTF-8 编译选项
+if(MSVC)
+    add_compile_options(/utf-8)
+endif()
+
+# 头文件目录
+set(INCLUDE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/include)
+
+# ============================
+# 主程序
+# ============================
+add_executable(dataflow_main
+    src/main.cpp
+    src/app.cpp
+)
+target_include_directories(dataflow_main PRIVATE ${INCLUDE_DIR})
+
+# ============================
+# 单元测试
+# ============================
+add_executable(basic_test
+    tests/basic_test.cpp
+    src/app.cpp
+)
+target_include_directories(basic_test PRIVATE ${INCLUDE_DIR})
+
+# 测试目标也可以通过 ctest 运行
+enable_testing()
+add_test(NAME basic_test COMMAND basic_test)

README.md

@@ -1,3 +1,73 @@
-# 测试
+# 数据处理与交互平台（测试系统）
 
-暂无描述
+## 概述
+
+本工程是一个模拟的数据处理与交互平台（测试系统），采用管道流式处理架构。
+核心功能包括：
+
+- **CSU-01 (DataReceiver)**：模拟 TCP 服务端，按自定义协议（消息头 + 消息体）接收外部数据源的数据。
+- **CSU-02 (DataProcessor)**：从队列获取原始数据，进行校验（非空、长度、字符合法性）和格式转换。
+- **CSU-03 (DataStorage)**：将处理后的数据持久化存储（当前版本使用内存向量模拟数据库写入）。
+
+三个模块通过线程安全队列（`ThreadSafeQueue`）解耦，形成数据流水线。
+
+## 工程结构
+
+```
+.
+├── CMakeLists.txt          # CMake 构建配置
+├── README.md               # 本文件
+├── include/
+│   └── app.hpp             # 公共头文件（类声明、数据结构）
+├── src/
+│   ├── main.cpp            # 命令行入口（完整流水线演示）
+│   └── app.cpp             # 模块实现
+└── tests/
+    └── basic_test.cpp      # 单元测试与集成测试
+```
+
+## 编译与运行
+
+### 前置条件
+
+- CMake >= 3.14
+- 支持 C++17 的编译器（GCC 8+, Clang 7+, MSVC 2019+）
+
+### 编译步骤
+
+```bash
+cd codegen-runs/codegen_fa54ba0e8c734a7683962ff923b15d4b
+mkdir -p build && cd build
+cmake ..
+cmake --build .
+```
+
+### 运行主程序
+
+```bash
+./dataflow_main
+```
+
+### 运行测试
+
+```bash
+./basic_test
+# 或
+ctest
+```
+
+## 模块映射
+
+| 文档中的软件单元 | 本工程类名 | 职责 |
+|:---|:---|:---|
+| CSU-01 | `DataReceiver` | 数据接收（模拟 TCP 服务端、协议拆包） |
+| CSU-02 | `DataProcessor` | 数据处理（校验、格式转换） |
+| CSU-03 | `DataStorage` | 数据存储（持久化到内存数据库） |
+| 内部接口 IF_CSU01_02 | `ThreadSafeQueue<std::string>` | 原始数据传递队列 |
+| 内部接口 CSU-02→CSU-03 | `ThreadSafeQueue<std::string>` | 处理后数据传递队列 |
+
+## 需求追踪
+
+| 需求标识 | 需求名称 | 对应函数 |
+|:---|:---|:---|
+| SRS-测试_F-001 | 数据接收功能 | `DataReceiver::start()`, `DataReceiver::receiveLoop()`, `DataReceiver::parsePacket()`, `DataProcessor::start()`, `DataProcessor::processLoop()`, `DataProcessor::validateAndTransform()`, `DataStorage::start()`, `DataStorage::storageLoop()`, `DataStorage::storeToDatabase()`, `main()` |

include/app.hpp

@@ -0,0 +1,341 @@
+#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

src/app.cpp

@@ -0,0 +1,359 @@
+#include "app.hpp"
+#include <iostream>
+#include <cstring>
+#include <sstream>
+#include <iomanip>
+#include <algorithm>
+
+namespace dataflow {
+
+// ============================================================================
+// DataReceiver
+// ============================================================================
+
+DataReceiver::DataReceiver(ThreadSafeQueue<std::string>& outputQueue)
+    : outputQueue_(outputQueue) {
+}
+
+DataReceiver::~DataReceiver() {
+    stop();
+}
+
+void DataReceiver::start() {
+    if (running_.exchange(true)) {
+        return;
+    }
+    workerThread_ = std::thread(&DataReceiver::receiveLoop, this);
+}
+
+void DataReceiver::stop() {
+    running_ = false;
+    if (workerThread_.joinable()) {
+        workerThread_.join();
+    }
+}
+
+void DataReceiver::setMockDataSource(const std::vector<std::string>& data) {
+    mockData_ = data;
+}
+
+size_t DataReceiver::getReceivedCount() const {
+    return receivedCount_.load();
+}
+
+size_t DataReceiver::getErrorCount() const {
+    return errorCount_.load();
+}
+
+std::string DataReceiver::getLastError() const {
+    std::lock_guard<std::mutex> lock(errorMutex_);
+    return lastError_;
+}
+
+/**
+ * @brief 内部工作线程主循环。
+ *
+ * 遍历模拟数据源，构造完整数据包并调用 parsePacket 解析，
+ * 将解析后的原始数据推入输出队列。
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：接收网络数据，拆包，传递给数据处理模块。
+ */
+void DataReceiver::receiveLoop() {
+    for (const auto& msgBody : mockData_) {
+        if (!running_) {
+            break;
+        }
+
+        // 构造完整数据包：4 字节长度前缀（小端序）+ 消息体
+        uint32_t bodyLen = static_cast<uint32_t>(msgBody.size());
+        std::vector<char> packet(4 + bodyLen);
+        std::memcpy(packet.data(), &bodyLen, sizeof(bodyLen));
+        if (bodyLen > 0) {
+            std::memcpy(packet.data() + 4, msgBody.data(), bodyLen);
+        }
+
+        // 解析数据包
+        size_t offset = 0;
+        std::string parsedData;
+        if (parsePacket(packet, offset, parsedData)) {
+            outputQueue_.push(parsedData);
+            receivedCount_++;
+        } else {
+            errorCount_++;
+            {
+                std::lock_guard<std::mutex> lock(errorMutex_);
+                lastError_ = "DataReceiver: failed to parse packet";
+            }
+        }
+
+        // 模拟间隔（每秒最多 100 次 => 每次至少 10ms）
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+}
+
+/**
+ * @brief 解析数据包。
+ * @param buffer  包含完整数据包（消息头 + 消息体）的字节缓冲区。
+ * @param offset  当前解析的起始偏移量（传入/传出）。
+ * @param outData [out] 解析出的消息体字符串。
+ * @return true 解析成功；false 数据不完整或格式错误。
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：根据约定的协议（消息头+消息体）进行拆包，解决粘包问题。
+ */
+bool DataReceiver::parsePacket(const std::vector<char>& buffer,
+                                size_t& offset,
+                                std::string& outData) {
+    // 检查剩余数据是否足够读取消息头
+    if (offset + 4 > buffer.size()) {
+        return false;
+    }
+
+    // 读取消息头中的长度字段（小端序）
+    uint32_t bodyLen = 0;
+    std::memcpy(&bodyLen, buffer.data() + offset, sizeof(bodyLen));
+    offset += 4;
+
+    // 检查消息体长度合法性
+    if (bodyLen > 1024) {
+        return false;
+    }
+
+    // 检查剩余数据是否足够读取完整消息体
+    if (offset + bodyLen > buffer.size()) {
+        return false;
+    }
+
+    // 读取消息体
+    outData.assign(buffer.data() + offset, bodyLen);
+    offset += bodyLen;
+
+    return true;
+}
+
+// ============================================================================
+// DataProcessor
+// ============================================================================
+
+DataProcessor::DataProcessor(ThreadSafeQueue<std::string>& inputQueue,
+                             ThreadSafeQueue<std::string>& outputQueue)
+    : inputQueue_(inputQueue), outputQueue_(outputQueue) {
+}
+
+DataProcessor::~DataProcessor() {
+    stop();
+}
+
+void DataProcessor::start() {
+    if (running_.exchange(true)) {
+        return;
+    }
+    workerThread_ = std::thread(&DataProcessor::processLoop, this);
+}
+
+void DataProcessor::stop() {
+    running_ = false;
+    if (workerThread_.joinable()) {
+        workerThread_.join();
+    }
+}
+
+size_t DataProcessor::getProcessedCount() const {
+    return processedCount_.load();
+}
+
+size_t DataProcessor::getDiscardedCount() const {
+    return discardedCount_.load();
+}
+
+size_t DataProcessor::getErrorCount() const {
+    return errorCount_.load();
+}
+
+std::string DataProcessor::getLastError() const {
+    std::lock_guard<std::mutex> lock(errorMutex_);
+    return lastError_;
+}
+
+/**
+ * @brief 内部工作线程主循环。
+ *
+ * 从输入队列获取原始数据，调用 validateAndTransform 进行校验和转换，
+ * 通过后推入输出队列，否则丢弃并记录错误。
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：对数据进行解析、校验（例如完整性校验）和格式转换。
+ */
+void DataProcessor::processLoop() {
+    while (running_) {
+        std::string rawData;
+        // 尝试非阻塞获取，以便能及时响应停止信号
+        if (inputQueue_.tryPop(rawData)) {
+            std::string processedData;
+            if (validateAndTransform(rawData, processedData)) {
+                outputQueue_.push(processedData);
+                processedCount_++;
+            } else {
+                discardedCount_++;
+                {
+                    std::lock_guard<std::mutex> lock(errorMutex_);
+                    lastError_ = "DataProcessor: data validation failed, discarded";
+                }
+            }
+        } else {
+            // 队列为空时短暂休眠
+            std::this_thread::sleep_for(std::chrono::milliseconds(5));
+        }
+    }
+}
+
+/**
+ * @brief 校验并转换原始数据。
+ * @param rawData      待校验的原始数据字符串。
+ * @param outProcessed [out] 校验通过后的处理结果。
+ * @return true 校验通过；false 校验失败。
+ *
+ * 校验规则：
+ * - 字符串非空
+ * - 长度不超过 1024 字符
+ * - 仅包含可打印 UTF-8 字符（简单 ASCII 可打印范围检查）
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：校验失败的数据丢弃，记录错误日志。
+ */
+bool DataProcessor::validateAndTransform(const std::string& rawData,
+                                          std::string& outProcessed) {
+    // 规则1：非空检查
+    if (rawData.empty()) {
+        return false;
+    }
+
+    // 规则2：长度不超过 1024 字符
+    if (rawData.size() > 1024) {
+        return false;
+    }
+
+    // 规则3：仅包含可打印 ASCII 字符的简单检查
+    // （用于模拟防止注入攻击的验证）
+    for (unsigned char ch : rawData) {
+        if (ch < 0x20 || ch > 0x7E) {
+            // 允许常见的空白字符（水平制表符、换行符等）用于实际业务场景，
+            // 但此处为简单校验，严格限定为可打印 ASCII。
+            // 对于中文 UTF-8 多字节字符，每个字节均大于 0x7E，
+            // 此简单校验不通过；实际系统中应使用完整的 UTF-8 验证。
+            // 本示例中，我们放宽限制以允许 UTF-8 中文数据通过。
+            // 放宽检查：仅拒绝控制字符（0x00 - 0x1F, 0x7F）
+            if (ch <= 0x1F || ch == 0x7F) {
+                return false;
+            }
+        }
+    }
+
+    // 转换：添加处理时间戳前缀（模拟格式转换）
+    auto now = std::chrono::system_clock::now();
+    auto nowT = std::chrono::system_clock::to_time_t(now);
+    std::stringstream ss;
+    ss << "[PROCESSED:" << std::put_time(std::localtime(&nowT), "%Y%m%d%H%M%S") << "] " << rawData;
+    outProcessed = ss.str();
+
+    return true;
+}
+
+// ============================================================================
+// DataStorage
+// ============================================================================
+
+DataStorage::DataStorage(ThreadSafeQueue<std::string>& inputQueue)
+    : inputQueue_(inputQueue) {
+}
+
+DataStorage::~DataStorage() {
+    stop();
+}
+
+void DataStorage::start() {
+    if (running_.exchange(true)) {
+        return;
+    }
+    workerThread_ = std::thread(&DataStorage::storageLoop, this);
+}
+
+void DataStorage::stop() {
+    running_ = false;
+    if (workerThread_.joinable()) {
+        workerThread_.join();
+    }
+}
+
+size_t DataStorage::getStoredCount() const {
+    return storedCount_.load();
+}
+
+size_t DataStorage::getErrorCount() const {
+    return errorCount_.load();
+}
+
+std::string DataStorage::getLastError() const {
+    std::lock_guard<std::mutex> lock(errorMutex_);
+    return lastError_;
+}
+
+std::vector<std::string> DataStorage::getAllStoredData() const {
+    std::lock_guard<std::mutex> lock(storageMutex_);
+    return storage_;
+}
+
+size_t DataStorage::getStorageSize() const {
+    std::lock_guard<std::mutex> lock(storageMutex_);
+    return storage_.size();
+}
+
+/**
+ * @brief 内部工作线程主循环。
+ *
+ * 从输入队列获取处理后的数据，调用 storeToDatabase 写入存储。
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：将处理后的数据持久化写入数据库。
+ */
+void DataStorage::storageLoop() {
+    while (running_) {
+        std::string data;
+        if (inputQueue_.tryPop(data)) {
+            if (storeToDatabase(data)) {
+                storedCount_++;
+            } else {
+                errorCount_++;
+                {
+                    std::lock_guard<std::mutex> lock(errorMutex_);
+                    lastError_ = "DataStorage: failed to store data";
+                }
+            }
+        } else {
+            std::this_thread::sleep_for(std::chrono::milliseconds(5));
+        }
+    }
+}
+
+/**
+ * @brief 将数据写入存储。
+ * @param data 需要存储的数据字符串。
+ * @return true 写入成功；false 写入失败。
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：处理数据库写入失败等异常，并记录日志。
+ */
+bool DataStorage::storeToDatabase(const std::string& data) {
+    try {
+        std::lock_guard<std::mutex> lock(storageMutex_);
+        storage_.push_back(data);
+        return true;
+    } catch (const std::exception& e) {
+        std::lock_guard<std::mutex> lock(errorMutex_);
+        lastError_ = std::string("DataStorage::storeToDatabase exception: ") + e.what();
+        return false;
+    }
+}
+
+} // namespace dataflow

src/main.cpp

@@ -0,0 +1,93 @@
+#include "app.hpp"
+#include <iostream>
+#include <vector>
+#include <string>
+#include <thread>
+#include <chrono>
+
+/**
+ * @brief 命令行入口。
+ *
+ * 构建完整的管道流式处理系统：
+ * 数据接收（CSU-01）→ 数据处理（CSU-02）→ 数据存储（CSU-03）。
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：系统整体功能验证。
+ */
+int main() {
+    std::cout << "=== 测试系统 - 管道流水线处理演示 ===" << std::endl;
+
+    // 创建模块间数据传递队列
+    dataflow::ThreadSafeQueue<std::string> queue01to02; // CSU-01 → CSU-02
+    dataflow::ThreadSafeQueue<std::string> queue02to03; // CSU-02 → CSU-03
+
+    // 创建三个处理模块
+    dataflow::DataReceiver  receiver(queue01to02);
+    dataflow::DataProcessor processor(queue01to02, queue02to03);
+    dataflow::DataStorage   storage(queue02to03);
+
+    // 设置模拟数据源（模拟外部数据源发送的数据）
+    std::vector<std::string> mockData = {
+        "Hello, system test!",
+        "Sensor reading: temperature=23.5, humidity=65",
+        "Event: system startup completed",
+        "",                    // 空数据，应被丢弃
+        "Normal data packet #4",
+        "中文测试数据包",
+        "Another measurement: pressure=1013.25",
+    };
+    receiver.setMockDataSource(mockData);
+
+    // 启动所有模块
+    std::cout << "启动各处理模块..." << std::endl;
+    receiver.start();
+    processor.start();
+    storage.start();
+
+    // 等待处理完成（模拟数据源发送完毕后，等待处理线程完成）
+    // 实际系统中会持续运行，此处等待足够时间让所有数据被处理
+    std::this_thread::sleep_for(std::chrono::milliseconds(200));
+
+    // 停止所有模块
+    std::cout << "停止各处理模块..." << std::endl;
+    receiver.stop();
+    processor.stop();
+    storage.stop();
+
+    // 输出统计信息
+    std::cout << "\n=== 处理统计 ===" << std::endl;
+    std::cout << "接收模块 (CSU-01):" << std::endl;
+    std::cout << "  成功接收: " << receiver.getReceivedCount() << " 个数据包" << std::endl;
+    std::cout << "  解析错误: " << receiver.getErrorCount() << " 个数据包" << std::endl;
+    if (receiver.getErrorCount() > 0) {
+        std::cout << "  最近错误: " << receiver.getLastError() << std::endl;
+    }
+
+    std::cout << "\n处理模块 (CSU-02):" << std::endl;
+    std::cout << "  成功处理: " << processor.getProcessedCount() << " 个数据包" << std::endl;
+    std::cout << "  被丢弃:   " << processor.getDiscardedCount() << " 个数据包" << std::endl;
+    if (processor.getDiscardedCount() > 0) {
+        std::cout << "  最近错误: " << processor.getLastError() << std::endl;
+    }
+
+    std::cout << "\n存储模块 (CSU-03):" << std::endl;
+    std::cout << "  成功存储: " << storage.getStoredCount() << " 条记录" << std::endl;
+    std::cout << "  存储错误: " << storage.getErrorCount() << " 条记录" << std::endl;
+    if (storage.getErrorCount() > 0) {
+        std::cout << "  最近错误: " << storage.getLastError() << std::endl;
+    }
+
+    // 输出存储的所有数据
+    std::cout << "\n=== 已存储的数据内容 ===" << std::endl;
+    auto storedData = storage.getAllStoredData();
+    for (size_t i = 0; i < storedData.size(); ++i) {
+        std::cout << "  [" << (i + 1) << "] " << storedData[i] << std::endl;
+    }
+
+    std::cout << "\n=== 演示结束 ===" << std::endl;
+
+    // 返回 0 表示成功，非 0 表示有错误
+    bool hasErrors = (receiver.getErrorCount() > 0)
+                     || (storage.getErrorCount() > 0);
+    return hasErrors ? 1 : 0;
+}

tests/basic_test.cpp

@@ -0,0 +1,219 @@
+#include "app.hpp"
+#include <cassert>
+#include <iostream>
+#include <thread>
+#include <chrono>
+#include <string>
+
+/**
+ * @brief 测试 CSU-01 DataReceiver 的数据包解析功能。
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：验证按协议拆包、异常日志等功能。
+ */
+void testDataReceiverParsing() {
+    std::cout << "[单元测试] testDataReceiverParsing ... ";
+
+    dataflow::ThreadSafeQueue<std::string> outputQueue;
+    dataflow::DataReceiver receiver(outputQueue);
+
+    std::vector<std::string> mockData = {
+        "Hello",
+        "World"
+    };
+    receiver.setMockDataSource(mockData);
+    receiver.start();
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    receiver.stop();
+
+    assert(receiver.getReceivedCount() == 2);
+    assert(receiver.getErrorCount() == 0);
+
+    std::string data1, data2;
+    assert(outputQueue.tryPop(data1));
+    assert(outputQueue.tryPop(data2));
+    assert(!outputQueue.tryPop(data1)); // 队列应为空
+
+    std::cout << "PASS" << std::endl;
+}
+
+/**
+ * @brief 测试 CSU-02 DataProcessor 的数据校验功能。
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：验证解析校验和错误处理逻辑。
+ */
+void testDataProcessorValidation() {
+    std::cout << "[单元测试] testDataProcessorValidation ... ";
+
+    dataflow::ThreadSafeQueue<std::string> inputQueue;
+    dataflow::ThreadSafeQueue<std::string> outputQueue;
+    dataflow::DataProcessor processor(inputQueue, outputQueue);
+
+    processor.start();
+
+    // 推送有效数据
+    inputQueue.push("Valid data packet");
+    inputQueue.push("Another valid data");
+
+    // 推送空数据（应被丢弃）
+    inputQueue.push("");
+
+    // 推送超长数据（应被丢弃）
+    std::string longData(1025, 'A');
+    inputQueue.push(longData);
+
+    // 等待处理
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    processor.stop();
+
+    // 验证统计
+    assert(processor.getProcessedCount() == 2);
+    assert(processor.getDiscardedCount() == 2);
+    assert(processor.getErrorCount() == 0);
+
+    // 验证输出队列内容
+    std::string processed;
+    assert(outputQueue.tryPop(processed));
+    assert(processed.find("[PROCESSED:") == 0);
+    assert(processed.find("Valid data packet") != std::string::npos);
+
+    assert(outputQueue.tryPop(processed));
+    assert(processed.find("Another valid data") != std::string::npos);
+
+    std::cout << "PASS" << std::endl;
+}
+
+/**
+ * @brief 测试 CSU-03 DataStorage 的数据存储功能。
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：验证数据入库和异常处理逻辑。
+ */
+void testDataStorage() {
+    std::cout << "[单元测试] testDataStorage ... ";
+
+    dataflow::ThreadSafeQueue<std::string> inputQueue;
+    dataflow::DataStorage storage(inputQueue);
+
+    storage.start();
+
+    inputQueue.push("Record 1");
+    inputQueue.push("Record 2");
+    inputQueue.push("Record 3");
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    storage.stop();
+
+    assert(storage.getStoredCount() == 3);
+    assert(storage.getErrorCount() == 0);
+    assert(storage.getStorageSize() == 3);
+
+    auto allData = storage.getAllStoredData();
+    assert(allData.size() == 3);
+    assert(allData[0] == "Record 1");
+    assert(allData[1] == "Record 2");
+    assert(allData[2] == "Record 3");
+
+    std::cout << "PASS" << std::endl;
+}
+
+/**
+ * @brief 测试 ThreadSafeQueue 的基本功能。
+ */
+void testThreadSafeQueue() {
+    std::cout << "[单元测试] testThreadSafeQueue ... ";
+
+    dataflow::ThreadSafeQueue<int> queue;
+
+    assert(queue.empty());
+    assert(queue.size() == 0);
+
+    queue.push(10);
+    queue.push(20);
+    queue.push(30);
+
+    assert(!queue.empty());
+    assert(queue.size() == 3);
+
+    int val;
+    assert(queue.tryPop(val));
+    assert(val == 10);
+    assert(queue.size() == 2);
+
+    assert(queue.tryPop(val));
+    assert(val == 20);
+
+    assert(queue.tryPop(val));
+    assert(val == 30);
+
+    assert(!queue.tryPop(val));
+    assert(queue.empty());
+
+    // 测试 clear
+    queue.push(1);
+    queue.push(2);
+    queue.clear();
+    assert(queue.empty());
+
+    std::cout << "PASS" << std::endl;
+}
+
+/**
+ * @brief 测试完整管道：接收→处理→存储。
+ *
+ * @requirement(name="数据接收功能", id="SRS-测试_F-001")
+ * 规则：验证 CSU-01 到 CSU-03 的完整协同工作。
+ */
+void testFullPipeline() {
+    std::cout << "[集成测试] testFullPipeline ... ";
+
+    dataflow::ThreadSafeQueue<std::string> queue01to02;
+    dataflow::ThreadSafeQueue<std::string> queue02to03;
+
+    dataflow::DataReceiver  receiver(queue01to02);
+    dataflow::DataProcessor processor(queue01to02, queue02to03);
+    dataflow::DataStorage   storage(queue02to03);
+
+    std::vector<std::string> mockData = {
+        "Pipeline test data 1",
+        "Pipeline test data 2",
+        ""  // 空数据应被丢弃
+    };
+    receiver.setMockDataSource(mockData);
+
+    receiver.start();
+    processor.start();
+    storage.start();
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    receiver.stop();
+    processor.stop();
+    storage.stop();
+
+    assert(receiver.getReceivedCount() == 3);
+    assert(processor.getProcessedCount() == 2);
+    assert(processor.getDiscardedCount() == 1);
+    assert(storage.getStoredCount() == 2);
+
+    auto stored = storage.getAllStoredData();
+    assert(stored.size() == 2);
+    assert(stored[0].find("Pipeline test data 1") != std::string::npos);
+    assert(stored[1].find("Pipeline test data 2") != std::string::npos);
+
+    std::cout << "PASS" << std::endl;
+}
+
+int main() {
+    std::cout << "=== 测试套件执行开始 ===" << std::endl;
+
+    testThreadSafeQueue();
+    testDataReceiverParsing();
+    testDataProcessorValidation();
+    testDataStorage();
+    testFullPipeline();
+
+    std::cout << "\n=== 所有测试通过 ===  (0)" << std::endl;
+    return 0;
+}