From 0b3d9f6313df93b5d895ecf7723c5c5dfff4ea16 Mon Sep 17 00:00:00 2001
From: zhaohe <h_zhaohe@163.com>
Date: Mon, 13 Nov 2023 18:34:54 +0800
Subject: [PATCH] update

---
 .../iflytop_front_end_service.cpp                  | 51 +++++++++++-----------
 .../iflytop_front_end_service.hpp                  |  8 ++--
 2 files changed, 30 insertions(+), 29 deletions(-)

diff --git a/src/iflytop/components/iflytop_front_end_service/iflytop_front_end_service.cpp b/src/iflytop/components/iflytop_front_end_service/iflytop_front_end_service.cpp
index 7e7a2b0..f32586f 100644
--- a/src/iflytop/components/iflytop_front_end_service/iflytop_front_end_service.cpp
+++ b/src/iflytop/components/iflytop_front_end_service/iflytop_front_end_service.cpp
@@ -22,15 +22,12 @@ void IflytopFrontEndService::initialize(string ip) {
     auto ws = webSocket.lock();
     if (!ws) return;
     ws->setOnMessageCallback([this, webSocket, connectionState](const ix::WebSocketMessagePtr& msg) {
-      shared_ptr<ix::WebSocketMessage> messageSPtr = make_shared<ix::WebSocketMessage>(*msg);
-
-      m_work_queue->enQueue([this, webSocket, connectionState, messageSPtr]() {
-        try {
-          onMessageCallback(webSocket, connectionState, messageSPtr);
-        } catch (const std::exception& e) {
-          logger->error("catch exception,onMessageCallback error: {}", e.what());
-        }
-      });
+      // shared_ptr<ix::WebSocketMessage> messageSPtr = make_shared<ix::WebSocketMessage>(*msg);
+      try {
+        onMessageCallback(webSocket, connectionState, msg);
+      } catch (const std::exception& e) {
+        logger->error("catch exception,onMessageCallback error: {}", e.what());
+      }
     });
   });
 
@@ -109,8 +106,7 @@ void IflytopFrontEndService::sendReport(json& report) {
   }
 }
 
-void IflytopFrontEndService::onMessageCallback(weak_ptr<WebSocket> webSocket, shared_ptr<ConnectionState> connectionState,
-                                               shared_ptr<ix::WebSocketMessage> msg) {
+void IflytopFrontEndService::onMessageCallback(weak_ptr<WebSocket> webSocket, shared_ptr<ConnectionState> connectionState, const ix::WebSocketMessagePtr& msg) {
   if (msg->type == ix::WebSocketMessageType::Open) {
     logger->info("New connection");
     logger->info("id     : {}", connectionState->getId());
@@ -119,27 +115,32 @@ void IflytopFrontEndService::onMessageCallback(weak_ptr<WebSocket> webSocket, sh
     for (auto it : msg->openInfo.headers) {
       logger->info("   {}: {}", it.first, it.second);
     }
-    onConnect(webSocket, msg);
+
+    m_work_queue->enQueue([this, webSocket]() { onConnect(webSocket); });
+
   } else if (msg->type == ix::WebSocketMessageType::Message) {
     logger->debug("downlink msg: {}", msg->str);
-    try {
-      processRxMessage(webSocket, msg);
-    } catch (const std::exception& e) {
-      logger->error("catch exception,processRxMessage error: {}", e.what());
-    }
-    // onMessage(webSocket, msg);
+    string msgtext = msg->str;
+    m_work_queue->enQueue([this, webSocket, msgtext]() {
+      try {
+        processRxMessage(webSocket, msgtext);
+      } catch (const std::exception& e) {
+        logger->error("catch exception,processRxMessage error: {}", e.what());
+      }
+    });
+
   } else if (msg->type == ix::WebSocketMessageType::Close) {
     logger->info("{} Closed connection", msg->closeInfo.remote);
-    onDisconnect(webSocket, msg);
+    m_work_queue->enQueue([this, webSocket]() { onDisconnect(webSocket); });
   }
 }
-void IflytopFrontEndService::processRxMessage(weak_ptr<WebSocket> webSocket, shared_ptr<ix::WebSocketMessage> msg) {
-  if (msg->binary) {
-    logger->warn("receive binary message,ignore");
-    return;
-  }
+void IflytopFrontEndService::processRxMessage(weak_ptr<WebSocket> webSocket, string rxmsg) {
+  // if (msg->binary) {
+  //   logger->warn("receive binary message,ignore");
+  //   return;
+  // }
 
-  string msgtext = msg->str;
+  string msgtext = rxmsg;
   json   message;
 
   try {
diff --git a/src/iflytop/components/iflytop_front_end_service/iflytop_front_end_service.hpp b/src/iflytop/components/iflytop_front_end_service/iflytop_front_end_service.hpp
index 62295d3..34a7503 100644
--- a/src/iflytop/components/iflytop_front_end_service/iflytop_front_end_service.hpp
+++ b/src/iflytop/components/iflytop_front_end_service/iflytop_front_end_service.hpp
@@ -43,8 +43,8 @@ class IflytopFrontEndService : public enable_shared_from_this<IflytopFrontEndSer
 
  public:
   // nod::signal<void(weak_ptr<WebSocket> webSocket, const ix::WebSocketMessagePtr& msg)> onMessage;
-  nod::signal<void(weak_ptr<WebSocket> webSocket, shared_ptr<ix::WebSocketMessage> msg)> onConnect;
-  nod::signal<void(weak_ptr<WebSocket> webSocket, shared_ptr<ix::WebSocketMessage> msg)> onDisconnect;
+  nod::signal<void(weak_ptr<WebSocket> webSocket)> onConnect;
+  nod::signal<void(weak_ptr<WebSocket> webSocket)> onDisconnect;
 
   nod::signal<void(weak_ptr<WebSocket> webSocket, json& cmd, json& receipt)> onMessage;
 
@@ -71,7 +71,7 @@ class IflytopFrontEndService : public enable_shared_from_this<IflytopFrontEndSer
   void sendToUDP(struct sockaddr_in* from, const char* data, size_t len) { m_udp->sendto(from, data, len); }
 
  private:
-  void onMessageCallback(weak_ptr<WebSocket> webSocket, shared_ptr<ConnectionState> connectionState, shared_ptr<ix::WebSocketMessage> msg);
-  void processRxMessage(weak_ptr<WebSocket> webSocket, shared_ptr<ix::WebSocketMessage> msg);
+  void onMessageCallback(weak_ptr<WebSocket> webSocket, shared_ptr<ConnectionState> connectionState, const ix::WebSocketMessagePtr& msg);
+  void processRxMessage(weak_ptr<WebSocket> webSocket, string rxmsg);
 };
 }  // namespace iflytop
\ No newline at end of file