#include "TimeReplicator.h" #include "DeviceHubWindow.h" #include <iostream> #include <QDateTime> #include <QTimer> TimeReplicator::TimeReplicator(QObject *parent) : DeviceBase(parent) { this->devType = "09"; connect(&this->serialUtil, &QSerialPortUtil::dataRecieved, this, &TimeReplicator::dataReceivedHandler); connect(this, &TimeReplicator::sendDataToDraw, ((DeviceHubWindow *)this->parent())->timeRepForm, &TimeReplicatorForm::drawDeviceFrameOnForm); connect(this, &TimeReplicator::sendCommandToDisplay, ((DeviceHubWindow *)this->parent())->timeRepForm, &TimeReplicatorForm::displayDeviceCommandOnForm); connect(((DeviceHubWindow *)this->parent())->kafkaConsumer, &QKafkaConsumer::messageRecieved, this, &TimeReplicator::commandReceivedHandler); kafkaProducer.setBrokers(SettingConfig::getInstance().KAFKA_BROKERS); kafkaProducer.setTopic(SettingConfig::getInstance().KAFKA_DATA_TOPIC); kafkaProducer.createProducer(); this->protocol = DeviceProtocolBase::deviceProtocolFactory(devType); } TimeReplicator::~TimeReplicator() { disconnect(&this->serialUtil, &QSerialPortUtil::dataRecieved, this, &TimeReplicator::dataReceivedHandler); } void TimeReplicator::mockReceivData() { QByteArray buffer; // time replicator buffer.append("$2B21308-13 21010011231111111111111111faf0*").append("\r\n"); // buffer.append("$3C21308-13 21010052207111111111111111111111111000000005984*").append("\r\n"); this->dataReceivedHandler(buffer); // QTimer::singleShot(1000, this, [=](){ // QByteArray buffer; // buffer.clear(); // buffer.append("$3C21308-13 21010052207111111111111111111111101000000005984*").append("\r\n"); // this->dataReceivedHandler(buffer); // }); // buffer.append("$3C21308-13 2101008220322222222111111110000000000000000ca24*"); // buffer.append("$3C21308-13 21010052207111111111111111111111111000000005984*").append("\r\n"); // buffer.append("$3C21308-13 2101005121511111111111111111111111111111111bcfe*").append("\r\n"); // buffer.append("$3E21308-13 2101001211308-13 2101001210929H.1.00S.1.00000292b*").append("\r\n"); } void TimeReplicator::mockReceivCmdReply() { QByteArray buffer; // time replicator buffer.append("$1941308-13 21010011eb4a*").append("\r\n"); this->dataReceivedHandler(buffer); } void TimeReplicator::dataReceivedHandler(QByteArray data) { this->dataBuff.append(data); QList<QByteArray> frameList = protocol->extractFrameList(this->dataBuff); if (frameList.size() > 0) { this->dataBuff.clear(); frameParse(frameList); } } void TimeReplicator::frameParse(QList<QByteArray> frameList) { for (int i = 0; i < frameList.size(); i++) { QByteArray frameByte = frameList.at(i); int frameType = protocol->checkFrame(frameByte); DeviceFrameBaseDto * frameDto = protocol->frameFactory(frameType); if (frameDto != nullptr) { // ★解析成数据对象 bool parse = protocol->parseDeviceFrameData(frameByte, frameDto, frameType); // 解析成功 if (parse == true) { QDateTime now = QDateTime::currentDateTime(); frameDto->timestamp = now.toString("yyyy-MM-dd HH:mm:ss.zzz"); frameDto->milisecond = now.toMSecsSinceEpoch(); frameDto->rawFrame = frameByte; frameDto->devCode = devCode; if (frameType == TimeReplicatorProtocolBM::TIME_REPLICATOR_FRAME_TYPE::STATUS_FRAME) { this->afterFrameParse(frameDto); } if (frameType == TimeReplicatorProtocolBM::TIME_REPLICATOR_FRAME_TYPE::CMDREP_FRAME) { frameDto->rawCommand = rawCommandBytes; frameDto->commandId = commandId; this->afterCommandReply(frameDto); } } // 在此处释放内存,不影响后续显示 // 不在此处释放内存则会导致内存持续增加 // 具体原因不明 delete frameDto; } } } void TimeReplicator::afterFrameParse(DeviceFrameBaseDto * frameDto) { // 0. 输出到日志文件中 QString date = frameDto->timestamp.mid(0, 10); // 1. 原始字节数组数据 QString filename = "raw_" + devCode + ".log"; QString content = frameDto->timestamp + " [recv] " + frameDto->rawFrame; QLogUtil::writeRawDataLogByDate(date, filename, content); // 2. 解析后的json数据 QString frameFilename = "frame_" + devCode + ".log"; QString frameContent = frameDto->timestamp + " [recv] " + QJsonDocument(frameDto->toJSON()).toJson(QJsonDocument::Compact); QLogUtil::writeChannelDataLogByDate(date, frameFilename, frameContent); // 3. 输出到中间件,执行后续处理过程 if (SettingConfig::getInstance().NEED_KAFKA == 1) { QJsonObject jsonObj = frameDto->toJSON(); jsonObj.insert("clientId", SettingConfig::getInstance().CLIENT_ID); jsonObj.insert("deviceId", deviceId); // 每10秒发送一次设备状态信息 if (QDateTime::currentDateTime().time().second() % 10 == 0) { kafkaProducer.produceMessage(QString(QJsonDocument(jsonObj).toJson(QJsonDocument::Compact))); } } // 4. 在界面上简单显示相差数据结果 emit this->sendDataToDraw(frameDto); } void TimeReplicator::afterCommandReply(DeviceFrameBaseDto * frameDto) { // 0. 输出到日志文件中 QString date = frameDto->timestamp.mid(0, 10); // 1. 原始字节数组数据 QString filename = "raw_" + devCode + ".log"; QString content = frameDto->timestamp + " [recv] " + frameDto->rawFrame.left(frameDto->rawFrame.size() - FRAME_TAIL.size()); QLogUtil::writeRawDataLogByDate(date, filename, content); // 2. 解析后的json数据 QString frameFilename = "frame_" + devCode + ".log"; QString frameContent = frameDto->timestamp + " [reply] " + QJsonDocument(frameDto->toJSON()).toJson(QJsonDocument::Compact); QLogUtil::writeChannelDataLogByDate(date, frameFilename, frameContent); // 3. 输出到中间件,执行后续处理过程 if (SettingConfig::getInstance().NEED_KAFKA == 1) { QJsonObject jsonObj = frameDto->toJSON(); kafkaProducer.produceMessage(SettingConfig::getInstance().KAFKA_CMDCB_TOPIC, QString(QJsonDocument(jsonObj).toJson(QJsonDocument::Compact))); } } void TimeReplicator::sendDataToSerial(QByteArray data) { data.append(FRAME_TAIL); if (SettingConfig::getInstance().WORK_MODE == "real") { if (serialUtil.isOpen() == true) { this->serialUtil.sendData(data); } else { CommandReplyDto replyDto; replyDto.devCode = devCode; replyDto.commandId = commandId; replyDto.rawCommand = rawCommandBytes; replyDto.cmdStatus = 0; replyDto.remarks = "serialport not avilable"; // 3. 输出到中间件,执行后续处理过程 if (SettingConfig::getInstance().NEED_KAFKA == 1) { kafkaProducer.produceMessage(SettingConfig::getInstance().KAFKA_CMDCB_TOPIC, QString(QJsonDocument(replyDto.toJSON()).toJson(QJsonDocument::Compact))); } return; } } // 记录日志 // 0. 输出到日志文件中 QDateTime now = QDateTime::currentDateTime(); QString date = now.toString("yyyy-MM-dd"); // 1. 原始字节数组数据 QString filename = "raw_" + devCode + ".log"; QString content = now.toString("yyyy-MM-dd HH:mm:ss.zzz") + " [send] " + data.left(data.size() - FRAME_TAIL.size()); QLogUtil::writeRawDataLogByDate(date, filename, content); std::cout << content.toStdString() << std::endl; } void TimeReplicator::commandReceivedHandler(QJsonObject command) { if (command.contains("deviceType") == false || command.value("deviceType").toString() != "09") { return; } if (command.contains("deviceId") == false || command.value("deviceId").toString() != deviceId) { return; } std::cout << QString(QJsonDocument(command).toJson(QJsonDocument::Compact)).toStdString() << std::endl; // 记录日志 // 0. 输出到日志文件中 QDateTime now = QDateTime::currentDateTime(); QString date = now.toString("yyyy-MM-dd"); // 1. 原始字节数组数据 QString filename = "frame_" + devCode + ".log"; QString content = now.toString("yyyy-MM-dd HH:mm:ss.zzz") + " [recv] " + command.find("cmdStr")->toString(); QLogUtil::writeChannelDataLogByDate(date, filename, content); QByteArray commandBytes = protocol->generateSettingCommand(devCode, command.value("command").toString(), command.value("params").toString()); this->sendDataToSerial(commandBytes); command.insert("rawCommand", QString::fromUtf8(commandBytes)); this->commandId = command.value("commandId").toString(); this->rawCommandBytes = commandBytes; // display on page emit sendCommandToDisplay(command); }