Fix live

wiliwili/include/api/live/danmaku_live.hpp

@@ -1,4 +1,3 @@
-
 //
 // Created by maye174 on 2023/4/6.
 //
@@ -26,25 +25,31 @@ class LiveDanmaku : public brls::Singleton<LiveDanmaku> {
  int uid;
  void connect(int room_id, int uid);
  void disconnect();
- bool is_connected();
-
  void send_join_request(int room_id, int uid);
+
  void send_heartbeat();
  void send_text_message(const std::string &message);
 
  void setonMessage(std::function<void(std::string)> func);
  std::function<void(std::string)> onMessage;
 
+ void set_wait_time(int time);
+ int wait_time = 600;
+
  LiveDanmaku();
  ~LiveDanmaku();
-private:
+
+ bool is_connected();
  std::atomic_bool connected{false};
+ bool is_evOK();
+ std::atomic_bool ms_ev_ok{false};
 
  std::thread mongoose_thread;
+ std::thread heartbeat_thread;
  std::mutex mongoose_mutex;
  mg_mgr *mgr;
  mg_connection *nc;
 
  //人气值
  int popularity = 0;
-};
+};

wiliwili/source/api/danmaku_live.cpp

@@ -8,6 +8,7 @@
 #include <cstdlib>
 #include <cstring>
 #include <iostream>
+#include <chrono>
 
 #include "live/danmaku_live.hpp"
 #include "live/ws_utils.hpp"
@@ -29,6 +30,14 @@ LiveDanmaku::LiveDanmaku() {
  printf("WSAStartup failed with error: %d\n", result);
  }
 #endif
+ heartbeat_thread = std::thread([this]() {
+ while (true) {
+ std::this_thread::sleep_for(std::chrono::seconds(20));
+ if (this->is_connected() and this->is_evOK()) {
+ this->send_heartbeat();
+ }
+ }
+ });
 }
 
 LiveDanmaku::~LiveDanmaku() {
@@ -47,19 +56,22 @@ void LiveDanmaku::connect(int room_id, int uid) {
  connected.store(true, std::memory_order_release);
 
  // Create and configure Mongoose connection
- struct mg_mgr *mgr = new mg_mgr;
+ this->mgr = new mg_mgr;
+ if(this->mgr == nullptr){
+ disconnect();
+ return;
+ }
+
  mg_log_set(MG_LL_NONE);
- mg_mgr_init(mgr);
- struct mg_connection *nc = mg_ws_connect(mgr, url.c_str(), mongoose_event_handler, this, nullptr);
+ mg_mgr_init(this->mgr);
 
- this->mgr = mgr;
- this->nc = nc;
+ this->nc = mg_ws_connect(this->mgr, url.c_str(), mongoose_event_handler, this, nullptr);
 
- if(nc == nullptr) {
+ if(this->nc == nullptr) {
  std::cout << "nc is null" << std::endl;
  disconnect();
  mg_mgr_free(this->mgr);
- delete mgr;
+ delete this->mgr;
  return;
  }
 
@@ -70,24 +82,13 @@ void LiveDanmaku::connect(int room_id, int uid) {
 
  // Start Mongoose event loop and heartbeat thread
  mongoose_thread = std::thread([this]() {
- int last = 0;
- int s = 0;
  while (this->is_connected()) {
  this->mongoose_mutex.lock();
  if(this->nc == nullptr) {
  break;
  }
  this->mongoose_mutex.unlock();
- mg_mgr_poll(this->mgr, 800);
- s += 1;
- if (s - last >= 36) {
- send_heartbeat();
- last = s;
- }
- if (s < 0) {
- s = 0;
- last = 0;
- }
+ mg_mgr_poll(this->mgr, wait_time);
  }
  mg_mgr_free(this->mgr);
  delete this->mgr;
@@ -110,10 +111,18 @@ void LiveDanmaku::disconnect() {
  }
 }
 
+void LiveDanmaku::set_wait_time(int time){
+ wait_time = time;
+}
+
 bool LiveDanmaku::is_connected() {
  return connected.load(std::memory_order_acquire);
 }
 
+bool LiveDanmaku::is_evOK(){
+ return ms_ev_ok.load(std::memory_order_acquire);
+}
+
 void LiveDanmaku::send_join_request(int room_id, int uid) {
  json join_request = {
  {"clientver", "1.6.3"},
@@ -127,6 +136,7 @@ void LiveDanmaku::send_join_request(int room_id, int uid) {
  std::vector<uint8_t> packet = encode_packet(0, 7, join_request_str);
  std::string packet_str(packet.begin(), packet.end());
  mongoose_mutex.lock();
+ if(this->nc == nullptr) return;
  mg_ws_send(this->nc, packet_str.data(), packet_str.size(), WEBSOCKET_OP_BINARY);
  mongoose_mutex.unlock();
 }
@@ -135,6 +145,7 @@ void LiveDanmaku::send_heartbeat() {
  std::vector<uint8_t> packet = encode_packet(0, 2, "");
  std::string packet_str(packet.begin(), packet.end());
  mongoose_mutex.lock();
+ if(this->nc == nullptr) return;
  mg_ws_send(this->nc, packet_str.data(), packet_str.size(), WEBSOCKET_OP_BINARY);
  mongoose_mutex.unlock();
 }
@@ -145,19 +156,22 @@ void LiveDanmaku::send_text_message(const std::string &message) {
 
 static void mongoose_event_handler(struct mg_connection *nc, int ev, void *ev_data, void *user_data) {
  LiveDanmaku *liveDanmaku = static_cast<LiveDanmaku *>(user_data);
+ liveDanmaku->ms_ev_ok.store(true, std::memory_order_release);
  if (ev == MG_EV_OPEN) {
  nc->is_hexdumping = 1;
  } else if (ev == MG_EV_ERROR) {
  //MG_ERROR(("%p %s", nc->fd, (char *) ev_data));
- liveDanmaku->disconnect();
+ //liveDanmaku->disconnect();
+ liveDanmaku->ms_ev_ok.store(false, std::memory_order_release);
  } else if (ev == MG_EV_WS_OPEN) {
  liveDanmaku->send_join_request(liveDanmaku->room_id, liveDanmaku->uid);
  } else if (ev == MG_EV_WS_MSG) {
  struct mg_ws_message *wm = (struct mg_ws_message *) ev_data;
  std::string message(wm->data.ptr, wm->data.len);
  liveDanmaku->onMessage(message);
  } else if(ev == MG_EV_CLOSE) {
- liveDanmaku->disconnect();
+ //liveDanmaku->disconnect();
+ liveDanmaku->ms_ev_ok.store(false, std::memory_order_release);
  }
 }