覆冰气象线程队列上传

app/src/main/cpp/CMakeLists.txt

@@ -392,6 +392,7 @@ add_library( # Sets the name of the library.
         #WeatherComm.cpp
         SensorsProtocol.cpp
         SerialComm.cpp
+        DataController.cpp
 
         ncnn/yolov5ncnn.cpp
 

app/src/main/cpp/DataController.cpp

@@ -0,0 +1,112 @@
+//
+// Created by shxy on 2025/3/12.
+//
+
+#include "DataController.h"
+
+DataController::DataController(CTerminal* pTerminal) :m_pTerminal(pTerminal), m_exit(false){}
+
+void DataController::Startup()
+{
+    m_thread = std::thread(DataThreadProc, this);
+}
+
+void DataController::DataThreadProc(DataController* pThis)
+{
+    pThis->DataProc();
+}
+void DataController::AddSendData(unsigned char frameType, unsigned char packetType, time_t timestamp,const vector<uint8_t> &data)
+{
+    DATA_INFO predata = {0};
+    predata.frameType = frameType;
+    predata.packetType = packetType;
+    predata.data = data;
+
+    m_locker.lock();
+    m_datas.push_back(predata);
+    m_locker.unlock();
+
+    m_sem.release();
+}
+
+bool DataController::WaitForResponse(unsigned char frameNo, unsigned char packetType, int sec)
+{
+    auto startTime = std::chrono::steady_clock::now();
+    while (true)
+    {
+        {
+            std::lock_guard<std::mutex> lock(m_responseLocker);
+            if (m_receivedresp.frameNo == frameNo && m_receivedresp.packetType == packetType)
+            {
+                m_receivedresp.frameNo = 0;
+                m_receivedresp.packetType = 0;
+                return true;
+            }
+        }
+
+        auto currentTime = std::chrono::steady_clock::now();
+        auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(currentTime - startTime).count();
+        if (elapsed >= sec)
+        {
+            return false;
+        }
+
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    }
+
+}
+
+void DataController::OnResponseReceived(unsigned char frameNo, unsigned char packetType)
+{
+    std::lock_guard<std::mutex> lock(m_responseLocker);
+    m_receivedresp.frameNo = frameNo;
+    m_receivedresp.packetType = packetType;
+}
+
+void DataController::DataProc()
+{
+    bool hasData;
+    DATA_INFO datasend;
+    string hispath = m_pTerminal->m_appPath + APP_DATA_DIR + "/" + APP_FILE_NAME_HIS_DB;
+    while(true) {
+        m_sem.acquire();
+
+        if (m_exit) {
+            break;
+        }
+
+        hasData = false;
+
+        m_locker.lock();
+        if (!m_datas.empty())
+        {
+            datasend = m_datas.front();
+            m_datas.pop_front();
+            hasData = true;
+        }
+        m_locker.unlock();
+
+        if (hasData) {
+            bool success = false;
+            for (int retry = 0; retry < 3; retry++)
+            {
+                unsigned char frameNo = 0;
+                m_pTerminal->DataSendTo(datasend.frameType, datasend.packetType, datasend.data, &frameNo);
+
+                if (WaitForResponse(frameNo, datasend.packetType, 8))
+                {
+                    success = true;
+                    UpdateHistoryStatus(hispath, datasend.packetType, datasend.timestamp);
+                    break;
+                }
+            }
+
+            if (!success)
+            {
+                std::lock_guard<std::mutex> lock(m_locker);
+                m_datas.push_back(datasend);
+            }
+        }
+
+    }
+}

app/src/main/cpp/DataController.h

@@ -0,0 +1,77 @@
+//
+// Created by shxy on 2025/3/12.
+//
+
+#ifndef MICROPHOTO_DATACONTROLLER_H
+#define MICROPHOTO_DATACONTROLLER_H
+
+
+#include <Buffer.h>
+#include <thread>
+#include <vector>
+#include <memory>
+#include <string>
+#include <mutex>
+#include <SemaphoreEx.h>
+#include <Client/Device.h>
+#include <Client/Terminal.h>
+#include <Client/Database.h>
+
+struct DATA_INFO
+{
+    unsigned char frameNo;
+    unsigned char frameType;
+    unsigned char packetType;
+    time_t timestamp;
+    vector<uint8_t> data;
+};
+
+struct DATA_RESP
+{
+    unsigned char frameNo;
+    unsigned char packetType;
+};
+
+class CTerminal;
+class DataController {
+public:
+    DataController(CTerminal *pTerminal);
+    ~DataController()
+    {
+        m_exit = true;
+        m_sem.release();
+        if (m_thread.joinable()) {
+            m_thread.join();
+        }
+    }
+
+    void Startup();
+    void AddSendData(unsigned char frameType, unsigned char packetType, time_t timestamp, const vector<uint8_t> &data);
+    void OnResponseReceived(unsigned char frameNo, unsigned char packetType);
+
+
+
+protected:
+    static void DataThreadProc(DataController* pThis);
+    void DataProc();
+    bool WaitForResponse(unsigned char frameNo, unsigned char packetType, int sec);
+
+
+
+protected:
+
+
+protected:
+    std::mutex m_locker;
+    std::mutex m_responseLocker;
+    CSemaphore m_sem;
+    std::deque<DATA_INFO> m_datas;
+    bool m_exit;
+    DATA_RESP m_receivedresp = {0,0};
+    std::thread m_thread;
+
+    CTerminal* m_pTerminal;
+};
+
+
+#endif //MICROPHOTO_DATACONTROLLER_H