Commit 1a66d846 authored by captainwong's avatar captainwong

update client&server

parent a50b0a0b
...@@ -36,12 +36,12 @@ struct OneTimeIniter { ...@@ -36,12 +36,12 @@ struct OneTimeIniter {
WSADATA wsa_data; WSADATA wsa_data;
WSAStartup(0x0201, &wsa_data); WSAStartup(0x0201, &wsa_data);
if (0 != evthread_use_windows_threads()) { if (0 != evthread_use_windows_threads()) {
fprintf(stderr, "failed to init libevent with thread by calling evthread_use_windows_threads\n"); JLOG_CRTC("failed to init libevent with thread by calling evthread_use_windows_threads");
abort(); abort();
} }
#else #else
if (0 != evthread_use_pthreads()) { if (0 != evthread_use_pthreads()) {
fprintf(stderr, "failed to init libevent with thread by calling evthread_use_pthreads\n"); JLOG_CRTC("failed to init libevent with thread by calling evthread_use_pthreads");
abort(); abort();
} }
#endif #endif
...@@ -74,7 +74,7 @@ struct Client::Impl ...@@ -74,7 +74,7 @@ struct Client::Impl
{ {
char buff[4096]; char buff[4096];
auto input = bufferevent_get_input(bev); auto input = bufferevent_get_input(bev);
JLOG_INFO("readcb, readable len {}", evbuffer_get_length(input)); JLOG_DBUG("readcb, readable len {}", evbuffer_get_length(input));
Client* client = (Client*)user_data; Client* client = (Client*)user_data;
if (client->userData_ && client->onMsg_) { if (client->userData_ && client->onMsg_) {
while (1) { while (1) {
...@@ -109,19 +109,24 @@ struct Client::Impl ...@@ -109,19 +109,24 @@ struct Client::Impl
static void eventcb(struct bufferevent* bev, short events, void* user_data) static void eventcb(struct bufferevent* bev, short events, void* user_data)
{ {
Client* client = (Client*)user_data; Client* client = (Client*)user_data;
JLOG_INFO("eventcb events={}", eventToString(events)); JLOG_DBUG("eventcb events={}", eventToString(events));
std::string msg; std::string msg;
if (events & BEV_EVENT_CONNECTED) { if (events & BEV_EVENT_CONNECTED) {
client->connected_ = true;
if (client->userData_ && client->onConn_) { if (client->userData_ && client->onConn_) {
client->onConn_(true, "connected", client->userData_); client->onConn_(true, "connected", client->userData_);
} }
client->lastTimeSendData = std::chrono::steady_clock::now(); client->lastTimeSendData = std::chrono::steady_clock::now();
if (client->strictTimer_) {
struct timeval tv = { client->timeout_, 0 };
event_add(event_new(client->impl_->base, -1, 0, Impl::timercb, client), &tv);
} else {
struct timeval tv = { 1, 0 }; struct timeval tv = { 1, 0 };
event_add(event_new(client->impl_->base, bufferevent_getfd(bev), 0, Impl::timercb, client), &tv); event_add(event_new(client->impl_->base, -1, 0, Impl::timercb, client), &tv);
}
return; return;
} else if (events & (BEV_EVENT_EOF)) { } else if (events & (BEV_EVENT_EOF)) {
msg = ("Connection closed"); msg = ("Connection closed");
...@@ -136,6 +141,8 @@ struct Client::Impl ...@@ -136,6 +141,8 @@ struct Client::Impl
} }
client->impl_->bev = nullptr; client->impl_->bev = nullptr;
client->connected_ = false;
if (client->userData_ && client->onConn_) { if (client->userData_ && client->onConn_) {
client->onConn_(false, msg, client->userData_); client->onConn_(false, msg, client->userData_);
} }
...@@ -147,9 +154,16 @@ struct Client::Impl ...@@ -147,9 +154,16 @@ struct Client::Impl
} }
} }
static void timercb(evutil_socket_t fd, short, void* user_data) static void timercb(evutil_socket_t, short, void* user_data)
{ {
Client* client = (Client*)user_data; Client* client = (Client*)user_data;
if (client->strictTimer_) {
if (client->userData_ && client->onTimer_) {
client->onTimer_(client->userData_);
}
struct timeval tv = { client->timeout_, 0 };
event_add(event_new(client->impl_->base, -1, 0, Impl::timercb, client), &tv);
} else {
auto now = std::chrono::steady_clock::now(); auto now = std::chrono::steady_clock::now();
auto diff = std::chrono::duration_cast<std::chrono::seconds>(now - client->lastTimeSendData); auto diff = std::chrono::duration_cast<std::chrono::seconds>(now - client->lastTimeSendData);
if (diff.count() >= client->timeout_) { if (diff.count() >= client->timeout_) {
...@@ -158,9 +172,10 @@ struct Client::Impl ...@@ -158,9 +172,10 @@ struct Client::Impl
} }
} }
struct timeval tv = { 1, 0 }; struct timeval tv = { 1, 0 };
client->impl_->timer = event_new(client->impl_->base, fd, 0, Impl::timercb, client); client->impl_->timer = event_new(client->impl_->base, -1, 0, Impl::timercb, client);
event_add(client->impl_->timer, &tv); event_add(client->impl_->timer, &tv);
} }
}
static void reconn_timercb(evutil_socket_t, short, void* user_data) static void reconn_timercb(evutil_socket_t, short, void* user_data)
{ {
...@@ -168,9 +183,9 @@ struct Client::Impl ...@@ -168,9 +183,9 @@ struct Client::Impl
do { do {
std::string msg = "Reconnecting " + client->impl_->ip + ":" + std::to_string(client->impl_->port); std::string msg = "Reconnecting " + client->impl_->ip + ":" + std::to_string(client->impl_->port);
if (client->userData_ && client->onConn_) { /*if (client->userData_ && client->onConn_) {
client->onConn_(false, msg, client->userData_); client->onConn_(false, msg, client->userData_);
} }*/
sockaddr_in sin = { 0 }; sockaddr_in sin = { 0 };
sin.sin_family = AF_INET; sin.sin_family = AF_INET;
...@@ -260,6 +275,11 @@ void Client::stop() ...@@ -260,6 +275,11 @@ void Client::stop()
auto old = autoReconnect_; auto old = autoReconnect_;
autoReconnect_ = false; autoReconnect_ = false;
if (impl_->timer) {
event_del(impl_->timer);
impl_->timer = nullptr;
}
if (impl_->bev) { if (impl_->bev) {
shutdown(bufferevent_getfd(impl_->bev), 1); shutdown(bufferevent_getfd(impl_->bev), 1);
//impl_->bev = nullptr; //impl_->bev = nullptr;
......
#pragma once #pragma once
# ifndef _CRT_SECURE_NO_WARNINGS # ifndef _CRT_SECURE_NO_WARNINGS
# define _CRT_SECURE_NO_WARNINGS # define _CRT_SECURE_NO_WARNINGS
...@@ -39,21 +39,27 @@ public: ...@@ -39,21 +39,27 @@ public:
void setUserData(void* d) { userData_ = d; } void setUserData(void* d) { userData_ = d; }
void setOnConnectionCallback(OnConnectinoCallback cb) { onConn_ = cb; } void setOnConnectionCallback(OnConnectinoCallback cb) { onConn_ = cb; }
void setOnMsgCallback(OnMessageCallback cb) { onMsg_ = cb; } void setOnMsgCallback(OnMessageCallback cb) { onMsg_ = cb; }
void setOnTimerCallback(OnTimerCallback cb, int seconds) { onTimer_ = cb; if (seconds > 0) { timeout_ = seconds; } } // strictTimer 是否严格以超时时间计算回调,否则会计算上次通信时间间隔
void setOnTimerCallback(OnTimerCallback cb, int seconds, bool strictTimer = false) {
onTimer_ = cb; if (seconds > 0) { timeout_ = seconds; } strictTimer_ = strictTimer;
}
void setAutoReconnect(bool b) { autoReconnect_ = b; } void setAutoReconnect(bool b) { autoReconnect_ = b; }
bool start(const std::string& ip, uint16_t port, std::string& msg); bool start(const std::string& ip, uint16_t port, std::string& msg);
void stop(); void stop();
void send(const char* data, size_t len); void send(const char* data, size_t len);
bool isStarted() const { return started_; } bool isStarted() const { return started_; }
bool isConnected() const { return connected_; }
protected: protected:
bool started_ = false; bool started_ = false;
bool connected_ = false;
bool autoReconnect_ = false; bool autoReconnect_ = false;
void* userData_ = nullptr; void* userData_ = nullptr;
OnConnectinoCallback onConn_ = nullptr; OnConnectinoCallback onConn_ = nullptr;
OnMessageCallback onMsg_ = nullptr; OnMessageCallback onMsg_ = nullptr;
OnTimerCallback onTimer_ = nullptr; OnTimerCallback onTimer_ = nullptr;
int timeout_ = 5; int timeout_ = 5;
bool strictTimer_ = false;
std::mutex mutex_ = {}; std::mutex mutex_ = {};
std::chrono::steady_clock::time_point lastTimeSendData = {}; std::chrono::steady_clock::time_point lastTimeSendData = {};
......
...@@ -371,7 +371,8 @@ void Server::stop() ...@@ -371,7 +371,8 @@ void Server::stop()
if (!impl) { return; } if (!impl) { return; }
if (impl->base) { if (impl->base) {
event_base_loopexit(impl->base, nullptr); timeval tv{ 0, 1000 };
event_base_loopexit(impl->base, &tv);
} }
if (impl->thread.joinable()) { if (impl->thread.joinable()) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment