提交 6c769d77 编写于 作者: Y youngwolf 提交者: youngowlf

Fix bug: the last msg will not be re-dispatched if on_msg_handle returned...

Fix bug: the last msg will not be re-dispatched if on_msg_handle returned false until new msgs come in.
Drop timer TIMER_SUSPEND_DISPATCH_MSG, it's useless.
上级 f645a52a
...@@ -151,9 +151,8 @@ protected: ...@@ -151,9 +151,8 @@ protected:
static const unsigned char TIMER_BEGIN = st_timer::TIMER_END; static const unsigned char TIMER_BEGIN = st_timer::TIMER_END;
static const unsigned char TIMER_DISPATCH_MSG = TIMER_BEGIN; static const unsigned char TIMER_DISPATCH_MSG = TIMER_BEGIN;
static const unsigned char TIMER_SUSPEND_DISPATCH_MSG = TIMER_BEGIN + 1; static const unsigned char TIMER_HANDLE_POST_BUFFER = TIMER_BEGIN + 1;
static const unsigned char TIMER_HANDLE_POST_BUFFER = TIMER_BEGIN + 2; static const unsigned char TIMER_RE_DISPATCH_MSG = TIMER_BEGIN + 2;
static const unsigned char TIMER_RE_DISPATCH_MSG = TIMER_BEGIN + 3;
static const unsigned char TIMER_END = TIMER_BEGIN + 10; static const unsigned char TIMER_END = TIMER_BEGIN + 10;
st_socket(boost::asio::io_service& io_service_) : st_timer(io_service_), _id(-1), next_layer_(io_service_), packer_(boost::make_shared<Packer>()), started_(false) {reset_state();} st_socket(boost::asio::io_service& io_service_) : st_timer(io_service_), _id(-1), next_layer_(io_service_), packer_(boost::make_shared<Packer>()), started_(false) {reset_state();}
...@@ -241,7 +240,6 @@ public: ...@@ -241,7 +240,6 @@ public:
void suspend_dispatch_msg(bool suspend) void suspend_dispatch_msg(bool suspend)
{ {
suspend_dispatch_msg_ = suspend; suspend_dispatch_msg_ = suspend;
stop_timer(TIMER_SUSPEND_DISPATCH_MSG);
do_dispatch_msg(true); do_dispatch_msg(true);
} }
bool suspend_dispatch_msg() const {return suspend_dispatch_msg_;} bool suspend_dispatch_msg() const {return suspend_dispatch_msg_;}
...@@ -348,9 +346,11 @@ protected: ...@@ -348,9 +346,11 @@ protected:
if (!temp_msg_buffer.empty()) if (!temp_msg_buffer.empty())
{ {
#ifndef ST_ASIO_FORCE_TO_USE_MSG_RECV_BUFFER #ifndef ST_ASIO_FORCE_TO_USE_MSG_RECV_BUFFER
if (!suspend_dispatch_msg_ && !posting)
{
out_container_type temp_2_msg_buffer; out_container_type temp_2_msg_buffer;
auto begin_time = statistic::local_time(); auto begin_time = statistic::local_time();
for (auto iter = std::begin(temp_msg_buffer); !suspend_dispatch_msg_ && !posting && iter != std::end(temp_msg_buffer);) for (auto iter = std::begin(temp_msg_buffer); iter != std::end(temp_msg_buffer);)
if (on_msg(*iter)) if (on_msg(*iter))
temp_msg_buffer.erase(iter++); temp_msg_buffer.erase(iter++);
else else
...@@ -366,6 +366,7 @@ protected: ...@@ -366,6 +366,7 @@ protected:
do_dispatch_msg(false); do_dispatch_msg(false);
} }
temp_msg_buffer.splice(std::begin(temp_msg_buffer), temp_2_msg_buffer); temp_msg_buffer.splice(std::begin(temp_msg_buffer), temp_2_msg_buffer);
}
#else #else
boost::unique_lock<boost::shared_mutex> lock(recv_msg_buffer_mutex); boost::unique_lock<boost::shared_mutex> lock(recv_msg_buffer_mutex);
if (splice_helper(recv_msg_buffer, temp_msg_buffer)) if (splice_helper(recv_msg_buffer, temp_msg_buffer))
...@@ -384,16 +385,12 @@ protected: ...@@ -384,16 +385,12 @@ protected:
void do_dispatch_msg(bool need_lock) void do_dispatch_msg(bool need_lock)
{ {
if (suspend_dispatch_msg_ || posting)
return;
boost::unique_lock<boost::shared_mutex> lock(recv_msg_buffer_mutex, boost::defer_lock); boost::unique_lock<boost::shared_mutex> lock(recv_msg_buffer_mutex, boost::defer_lock);
if (need_lock) lock.lock(); if (need_lock) lock.lock();
if (suspend_dispatch_msg_)
{
if (!dispatching && !recv_msg_buffer.empty())
set_timer(TIMER_SUSPEND_DISPATCH_MSG, 24 * 60 * 60 * 1000, [this](unsigned char id)->bool {return ST_THIS timer_handler(id);}); //one day
}
else if (!posting)
{
auto dispatch_all = false; auto dispatch_all = false;
if (stopped()) if (stopped())
dispatch_all = !(dispatching = false); dispatch_all = !(dispatching = false);
...@@ -401,16 +398,18 @@ protected: ...@@ -401,16 +398,18 @@ protected:
{ {
if (!started()) if (!started())
dispatch_all = true; dispatch_all = true;
else if (!recv_msg_buffer.empty()) else if (!last_dispatch_msg.empty())
{ {
dispatching = true; dispatching = true;
if (last_dispatch_msg.empty()) post([this]() {ST_THIS msg_handler();});
}
else if (!recv_msg_buffer.empty())
{ {
last_dispatch_msg.restart(recv_msg_buffer.front().begin_time); last_dispatch_msg.restart(recv_msg_buffer.front().begin_time);
last_dispatch_msg.swap(recv_msg_buffer.front()); last_dispatch_msg.swap(recv_msg_buffer.front());
recv_msg_buffer.pop_front(); recv_msg_buffer.pop_front();
}
dispatching = true;
post([this]() {ST_THIS msg_handler();}); post([this]() {ST_THIS msg_handler();});
} }
} }
...@@ -418,12 +417,14 @@ protected: ...@@ -418,12 +417,14 @@ protected:
if (dispatch_all) if (dispatch_all)
{ {
#ifndef ST_ASIO_DISCARD_MSG_WHEN_LINK_DOWN #ifndef ST_ASIO_DISCARD_MSG_WHEN_LINK_DOWN
if (!last_dispatch_msg.empty())
on_msg_handle(last_dispatch_msg, true);
st_asio_wrapper::do_something_to_all(recv_msg_buffer, [this](out_msg& msg) {ST_THIS on_msg_handle(msg, true);}); st_asio_wrapper::do_something_to_all(recv_msg_buffer, [this](out_msg& msg) {ST_THIS on_msg_handle(msg, true);});
#endif #endif
last_dispatch_msg.clear();
recv_msg_buffer.clear(); recv_msg_buffer.clear();
} }
} }
}
//must mutex send_msg_buffer before invoke this function //must mutex send_msg_buffer before invoke this function
bool do_direct_send_msg(InMsgType&& msg) bool do_direct_send_msg(InMsgType&& msg)
...@@ -464,9 +465,6 @@ private: ...@@ -464,9 +465,6 @@ private:
stat.recv_idle_sum += statistic::local_time() - recv_idle_begin_time; stat.recv_idle_sum += statistic::local_time() - recv_idle_begin_time;
dispatch_msg(); dispatch_msg();
break; break;
case TIMER_SUSPEND_DISPATCH_MSG: //suspend dispatching msgs
do_dispatch_msg(true);
break;
case TIMER_HANDLE_POST_BUFFER: case TIMER_HANDLE_POST_BUFFER:
{ {
boost::unique_lock<boost::shared_mutex> lock(post_msg_buffer_mutex); boost::unique_lock<boost::shared_mutex> lock(post_msg_buffer_mutex);
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册