LogicSystem#

1
#pragma once
2
#include "Const.h"
3
#include "Singleton.h"
4
#include <map>
5
#include <functional>
6
#include <mutex>
7
#include <queue>
8
#include <unordered_map>
9
class CSession;
10
class LogicNode;
11

12
typedef std::function<void(std::shared_ptr<CSession>, const short& msg_id, const std::string& msg_data)> node_callback;
13

14
class LogicSystem : public Singleton<LogicSystem>{
15
  friend class Singleton<LogicSystem>;
16
public:
17
  ~LogicSystem();
18
  void postMsgToQueue(std::shared_ptr<LogicNode> msg_node);
19
private:
20
  LogicSystem();
21
  void registerHanlder();
22
  void dealMsg();
23

24
  // 登录逻辑处理
25
  void loginHandler(std::shared_ptr<CSession> session, const short& msg_id, const std::string& msg_data);
26

27
  //
28
  bool _b_stop;
29

30
  // 单线程处理消息节点。
31
  std::mutex _node_mutex;
32
  std::queue<std::shared_ptr<LogicNode>> _node_queue;
33
  std::thread _work_thread;
34
  std::condition_variable _consume;
35

36
  // 不同消息的逻辑函数列表
37
  std::map <short, node_callback> _handlers;
38

39
  // 待处理的用户列表
40
  //std::unordered_map<int, std::shared_ptr<UserInfo>> _users;
41
};

1
#include "LogicSystem.h"
2
#include "CSession.h"
3

4
LogicSystem::LogicSystem() : _b_stop(false) {
5
  registerHanlder();
6
  _work_thread = std::thread(&LogicSystem::dealMsg, this);
7

8
}
9

10
LogicSystem::~LogicSystem() {
11
  std::cout << "LogicSystem destroy~" << std::endl;
12
  _b_stop = true;
13
  _consume.notify_one();
14
  _work_thread.join();
15
}
16

17
void LogicSystem::dealMsg() {
18
  std::shared_ptr<LogicNode> node;
19
  while (true) {
20
    std::unique_lock<std::mutex> lock(_node_mutex);
21
    while (_node_queue.empty() || !_b_stop) {
22
      _consume.wait(lock);
23
    }
24

25
    if (_b_stop) {
26
      while (!_node_queue.empty()) {
27
        auto node = _node_queue.front();
28
        auto msg_id = node->_recv_node->_msg_id;
29
        std::cout << "recv_msg id is " << node->_recv_node->_msg_id << std::endl;
30
        auto callback_iter = _handlers.find(msg_id);
31
        if (callback_iter == _handlers.end()) {
32
          _node_queue.pop();
33
          std::cout << "msg_id " << msg_id << " handler not found." << std::endl;
34
          continue;
35
        }
36
        callback_iter->second(node->_session, node->_recv_node->_msg_id,
37
          std::string(node->_recv_node->_data, node->_recv_node->_cur_length));
38
      }
39
      break;
40
    }
41

42
    auto node = _node_queue.front();
43
    auto msg_id = node->_recv_node->_msg_id;
44
    std::cout << "logicSystem msg_id is: " << msg_id << std::endl;
45
    auto callback_iter = _handlers.find(msg_id);
46
    if (callback_iter == _handlers.end()) {
47
      _node_queue.pop();
48
      std::cout << "msg_id " << msg_id << " handler not found." << std::endl;
49
      continue;
50
    }
51
    callback_iter->second(node->_session, node->_recv_node->_msg_id,
52
      std::string(node->_recv_node->_data, node->_recv_node->_cur_length));
53

54
    _node_queue.pop();
55
  }
56
}
57

58
void LogicSystem::postMsgToQueue(std::shared_ptr<LogicNode> node) {
59
  std::unique_lock<std::mutex> lock(_node_mutex);
60
  _node_queue.push(node);
61
  // 如果由空变为不空
62
  if (_node_queue.size() == 1) {
63
    lock.unlock();
64
    _consume.notify_one();
65
  }
66
}
67

68
void LogicSystem::registerHanlder()
69
{
70
  _handlers[MSG_CHAT_LOGIN] = std::bind(&LogicSystem::loginHandler, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
71
}
72

73
void LogicSystem::loginHandler(std::shared_ptr<CSession> session, const short& msg_id, const std::string& msg_data) {
74
  Json::Reader reader;
75
  Json::Value root;
76
  reader.parse(msg_data, root);
77
  std::cout << "user login uid is: " << root["uid"].asInt() <<
78
    ", user token is: " << root["token"].asString() << std::endl;
79

80
  // process data
81
  // ...
82

83
  // response data
84
  std::string return_str = root.toStyledString();
85
  session->send(return_str, msg_id);
86
}