update

src/iflytop/components/iflytop_front_end_service/iflytop_front_end_service.cpp

@@ -22,17 +22,14 @@ 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]() {
+      // shared_ptr<ix::WebSocketMessage> messageSPtr = make_shared<ix::WebSocketMessage>(*msg);
       try {
-          onMessageCallback(webSocket, connectionState, messageSPtr);
+        onMessageCallback(webSocket, connectionState, msg);
       } catch (const std::exception& e) {
         logger->error("catch exception,onMessageCallback error: {}", e.what());
       }
     });
   });
-  });
 
   m_report_server.reset(new WebSocketServer(19002, ip));
   m_report_server->setOnConnectionCallback([this](weak_ptr<WebSocket> webSocket, shared_ptr<ConnectionState> connectionState) {
@@ -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);
+    string msgtext = msg->str;
+    m_work_queue->enQueue([this, webSocket, msgtext]() {
       try {
-      processRxMessage(webSocket, msg);
+        processRxMessage(webSocket, msgtext);
       } catch (const std::exception& e) {
         logger->error("catch exception,processRxMessage error: {}", e.what());
       }
-    // onMessage(webSocket, msg);
+    });
+
   } 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 {

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