服务器参考乐外卖的更新

zhipuzi_pos_windows_server/helper/CAliyunMNS.cpp

@@ -57,6 +57,7 @@ std::string CAliyunMNS::getMessage()
 {
     if(is_init == false)
     {
+		LOG_INFO("is_init is false");
         return "error!";
     }
 

zhipuzi_pos_windows_server/network/CClientSession.cpp

@@ -23,6 +23,26 @@ void CClientSession::start()
  **/
 void CClientSession::stop()
 {
+	//先给客户端发送一个下线通知
+	rapidjson::Document doc;
+	doc.SetObject();
+	rapidjson::Document::AllocatorType& allocator = doc.GetAllocator(); //获取分配器
+
+	std::string timestamp = to_string(time(NULL));
+
+	doc.AddMember("msg_type", "0", allocator);
+	doc.AddMember("timestamp", StringRef(timestamp.c_str(), timestamp.length()), allocator);
+
+	rapidjson::StringBuffer buffer;
+	rapidjson::Writer<StringBuffer> writer(buffer);
+	doc.Accept(writer);
+
+	//返回给接入层的消息
+	std::string login_out_msg = buffer.GetString();
+
+	send_message(login_out_msg);
+
+	//然后关闭socket
 	socket_.close();
 
 	//注意这里不能调用delete this，因为后面会异步返回，要判断状态，如果这里delete了，后面就会内存越界

zhipuzi_pos_windows_server/network/CServer.cpp

@@ -54,6 +54,13 @@ void CServer::Init()
 		t.detach();
 	}
 
+	//开一个线程，每隔30分钟清理一次过期的消息
+	for (int i = 0; i < 1; i++)
+	{
+		std::thread t(&CServer::HandleCleanupDueMessage, this);
+		t.detach();
+	}
+
 	//开始异步执行
 	io_context_.run();
 }
@@ -74,8 +81,9 @@ void CServer::InitSqlLite()
 
 	m_rc = sqlite3_open(s_path.c_str(), &m_db);
 #else
-	LOG_INFO("absopath:" << CSystem::getAbsopath().c_str());
 	std::string db_path = CSystem::getAbsopath() + "../db/pos.db";
+	LOG_INFO("db path:" << db_path.c_str());
+
 	m_rc = sqlite3_open(db_path.c_str(), &m_db);
 #endif
 
@@ -160,11 +168,11 @@ bool CServer::AddMessageToDB(std::string username, std::string due_time, std::st
 
 	if (result == SQLITE_OK)
 	{
-		LOG_INFO("save params success");
+		//LOG_INFO("save params success");
 		return true;
 	}
 
-	LOG_INFO("save params fail");
+	//LOG_INFO("save params fail");
 	return false;
 }
 
@@ -234,9 +242,7 @@ void CServer::HandleOfflineMessage()
 					bool ret = session->send_message(data);
 					if (ret == false)
 					{
-						//发送失败，把数据写会数据库，等下次发送
-						AddMessageToDB(username, due_time, data);
-
+						//发送失败，等下次发送
 						break;
 					}
 
@@ -297,7 +303,7 @@ void CServer::BindUsername(std::string username, CClientSession* session)
 
 	if (m_clients_map.find(username) != m_clients_map.end() && m_clients_map[username]->GetNum() < session->GetNum())
 	{
-		//之前已经存在了一个，先把直接的关闭掉（这个是服务器端主动的关闭）
+		//之前已经存在了一个，先把之前的关闭掉（这个是服务器端主动的关闭）
 		m_clients_map[username]->stop();
 	}
 
@@ -331,15 +337,18 @@ void CServer::ReceiveMNSMessage()
         return;
     }
 
+	LOG_INFO("Init AliyunMNS success!");
+
     while(true)
     {
         std::string message = mns.getMessage();
         if(message == "error!")
         {
+			LOG_INFO("message error!");
             continue;
         }
 
-        LOG_INFO("get new message:" << message.c_str());
+        //LOG_INFO("get new message:" << message.c_str());
 
         //获取到了新的消息，开始进行处理
         rapidjson::Document document;
@@ -351,9 +360,23 @@ void CServer::ReceiveMNSMessage()
         }
 
         //处理消息类型
+		if (!document["username"].IsString() || !document["timestamp"].IsString())
+		{
+			continue;
+		}
+
+		std::string use_time;
+		if (document["use_time"].IsInt())
+		{
+			use_time = to_string(document["use_time"].GetInt());
+		}
+		else
+		{
+			use_time = document["use_time"].GetString();
+		}
+
 		std::string username = document["username"].GetString();
 		std::string timestamp = document["timestamp"].GetString();
-		std::string use_time = document["use_time"].GetString();
 
 		rapidjson::Value& data = document["data"];
 
@@ -384,8 +407,6 @@ void CServer::ReceiveMNSMessage()
 		m_queue_mutex.lock();
 		m_message_queue.push(newMessage);
 		m_queue_mutex.unlock();
-
-		//LOG_INFO("push message to queue!");
     }
 }
 
@@ -424,7 +445,7 @@ void CServer::SendMessageToClient()
 		{
 			m_map_mutex.unlock();
 
-			LOG_INFO("can not find client, save to db");
+			LOG_INFO("can not find client, save to db, username:"<<username.c_str());
 
 			//客户端不在线，操作存数据库
 			AddMessageToDB(username, due_time, data);
@@ -439,15 +460,29 @@ void CServer::SendMessageToClient()
 			bool ret = session->send_message(data);
 			if (ret == false)
 			{
-				LOG_INFO("send to client fail, save it to db");
+				LOG_INFO("send to client fail, save it to db, username:" << username.c_str());
 				//如果发送失败了，把消息存回到数据库
 				AddMessageToDB(username, due_time, data);
 			}
 			else
 			{
-				LOG_INFO("send to client success");
+				LOG_INFO("send to client success, username:" << username.c_str());
 			}
 		}
+	}	
+}
+
+/*
+ *每30分钟清理一次数据库
+ **/
+void CServer::HandleCleanupDueMessage()
+{
+	while (true)
+	{
+		std::string curTime = CLewaimaiTime::DatetimeToString(time(NULL));
+		std::string sql = "DELETE FROM pos_message WHERE due_time <= '" + curTime + "';";
+		sqlite3_exec(m_db, sql.c_str(), 0, 0, 0);
+
+		CSystem::my_sleep(60 * 30);
 	}
-	
 }

zhipuzi_pos_windows_server/network/CServer.h

@@ -36,6 +36,8 @@ public:
 
 	void HandleOfflineMessage();
 
+	void HandleCleanupDueMessage();
+
 private:
     void start_accept();