341 lines
10 KiB
C++
341 lines
10 KiB
C++
#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
|