server_common/f8
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
aozhiwei 2019-07-08 17:22:35 +08:00
parent 645ee0102f
commit a09180d909
5 changed files with 28 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cpp/dbpool.cc
View File

@ -319,8 +319,14 @@ namespace f8
void Init()
{
auto free_custom_msg = [] (const a8::XParams& params)
{
a8::XParams* param = (a8::XParams*)params.param1.GetUserData();
DataSet* data_set = (DataSet*)param->param2.GetUserData();
delete data_set;
};
curr_seqid = 1000001;
exec_async_query_msgid = MsgQueue::Instance()->AllocIMMsgId();
exec_async_query_msgid = MsgQueue::Instance()->AllocIMMsgId(free_custom_msg);
}
void UnInit()

2
cpp/httpclientpool.cc
View File

@ -276,7 +276,7 @@ namespace f8
void Init()
{
curr_seqid = 1000001;
exec_async_http_msgid = MsgQueue::Instance()->AllocIMMsgId();
exec_async_http_msgid = MsgQueue::Instance()->AllocIMMsgId(nullptr);
}
void UnInit()

17
cpp/msgqueue.cc
View File

@ -18,6 +18,7 @@ namespace f8
public:
int curr_im_msgid = 10000;
std::map<int, list_head> msg_handlers;
std::map<int, CustomIMMsgFreeFunc*> custom_free_funcs;
~MsgQueueImp()
{
@ -109,9 +110,21 @@ namespace f8
return imp_->RegisterCallBack(msgid, handle_func);
}
int MsgQueue::AllocIMMsgId()
int MsgQueue::AllocIMMsgId(CustomIMMsgFreeFunc free_func)
{
return ++imp_->curr_im_msgid;
int custom_im_msgid = ++imp_->curr_im_msgid;
if (free_func) {
imp_->custom_free_funcs[custom_im_msgid] = free_func;
}
return custom_im_msgid;
}
void MsgQueue::FreeCustomIMMsg(a8::XParams& param)
{
auto itr = imp_->custom_free_funcs.find(param.sender.GetInt());
if (itr != imp_->custom_free_funcs.end()) {
itr->second(param);
}
}
void MsgQueue::ProcessMsg(int msgid, const a8::XParams& param)

4
cpp/msgqueue.h
View File

@ -3,6 +3,7 @@
namespace f8
{
typedef std::function<void (const a8::XParams& param)> MsgHandleFunc;
typedef void CustomIMMsgFreeFunc(const a8::XParams& param);
typedef list_head* CallBackHandle;
class MsgQueueImp;
@ -21,7 +22,8 @@ namespace f8
void AddDelayMsg(int msgid, a8::XParams param, int milli_seconds);
void RemoveCallBack(CallBackHandle handle);
CallBackHandle RegisterCallBack(int msgid, MsgHandleFunc handle_func);
int AllocIMMsgId();
int AllocIMMsgId(CustomIMMsgFreeFunc free_func);
void FreeCustomIMMsg(a8::XParams& param);
void ProcessMsg(int msgid, const a8::XParams& param);
//线程安全版本

2
cpp/types.h
View File

@ -12,6 +12,8 @@ namespace f8
bool pending = false;
unsigned long saddr = 0;
int socket_handle = 0;
time_t create_time = 0;
time_t handle_time = 0;
std::string query_str;
a8::XObject request;
a8::MutableXObject* resp_xobj = nullptr;