Commit 0b41c6de authored by captainwong's avatar captainwong

update

parent 1616119c
...@@ -64,8 +64,7 @@ std::string file_name = "" ...@@ -64,8 +64,7 @@ std::string file_name = ""
fprintf(stderr, "Log initialization failed: %s\n", ex.what()); fprintf(stderr, "Log initialization failed: %s\n", ex.what());
#endif #endif
exit(0); exit(0);
} }
} }
......
...@@ -215,13 +215,14 @@ struct simple_libevent_client::Impl ...@@ -215,13 +215,14 @@ struct simple_libevent_client::Impl
} }
}; };
bool simple_libevent_client::start(const std::string& ip, uint16_t port, std::string& msg) bool simple_libevent_client::start(const std::string& ip, uint16_t port, std::string& msg, bool start_in_thread)
{ {
AUTO_LOG_FUNCTION; AUTO_LOG_FUNCTION;
do { do {
stop(); stop();
std::lock_guard<std::mutex> lg(mutex_); //std::lock_guard<std::mutex> lg(mutex_);
mutex_.lock();
impl_ = new Impl(this); impl_ = new Impl(this);
impl_->ip = ip; impl_->ip = ip;
...@@ -230,6 +231,7 @@ bool simple_libevent_client::start(const std::string& ip, uint16_t port, std::st ...@@ -230,6 +231,7 @@ bool simple_libevent_client::start(const std::string& ip, uint16_t port, std::st
impl_->base = event_base_new(); impl_->base = event_base_new();
if (!impl_->base) { if (!impl_->base) {
msg = "init libevent failed"; msg = "init libevent failed";
mutex_.unlock();
break; break;
} }
...@@ -241,6 +243,7 @@ bool simple_libevent_client::start(const std::string& ip, uint16_t port, std::st ...@@ -241,6 +243,7 @@ bool simple_libevent_client::start(const std::string& ip, uint16_t port, std::st
impl_->bev = bufferevent_socket_new(impl_->base, -1, BEV_OPT_CLOSE_ON_FREE); impl_->bev = bufferevent_socket_new(impl_->base, -1, BEV_OPT_CLOSE_ON_FREE);
if (!impl_->bev) { if (!impl_->bev) {
msg = ("allocate bufferevent failed"); msg = ("allocate bufferevent failed");
mutex_.unlock();
break; break;
} }
bufferevent_setcb(impl_->bev, Impl::readcb, Impl::writecb, Impl::eventcb, this); bufferevent_setcb(impl_->bev, Impl::readcb, Impl::writecb, Impl::eventcb, this);
...@@ -248,18 +251,26 @@ bool simple_libevent_client::start(const std::string& ip, uint16_t port, std::st ...@@ -248,18 +251,26 @@ bool simple_libevent_client::start(const std::string& ip, uint16_t port, std::st
if (bufferevent_socket_connect(impl_->bev, (sockaddr*)(&sin), sizeof(sin)) < 0) { if (bufferevent_socket_connect(impl_->bev, (sockaddr*)(&sin), sizeof(sin)) < 0) {
msg = ("error starting connection"); msg = ("error starting connection");
mutex_.unlock();
break; break;
} }
lastTimeSendData = std::chrono::steady_clock::now(); lastTimeSendData = std::chrono::steady_clock::now();
impl_->thread = std::thread([this]() { if (start_in_thread) {
JLOG_WARN("libevent thread started"); impl_->thread = std::thread([this]() {
JLOG_WARN("libevent thread started");
event_base_dispatch(this->impl_->base);
JLOG_WARN("libevent thread exited");
});
started_ = true;
mutex_.unlock();
} else {
started_ = true;
mutex_.unlock();
event_base_dispatch(this->impl_->base); event_base_dispatch(this->impl_->base);
JLOG_WARN("libevent thread exited"); }
});
started_ = true;
return true; return true;
} while (0); } while (0);
......
...@@ -55,8 +55,14 @@ public: ...@@ -55,8 +55,14 @@ public:
void setOnTimerCallback(OnTimerCallback cb, int seconds, bool strictTimer = false) { void setOnTimerCallback(OnTimerCallback cb, int seconds, bool strictTimer = false) {
onTimer_ = cb; if (seconds > 0) { timeout_ = seconds; } strictTimer_ = strictTimer; onTimer_ = cb; if (seconds > 0) { timeout_ = seconds; } strictTimer_ = strictTimer;
} }
// 设置生命周期长度,seconds 秒后退出工作循环/工作线程,设置 <=0 值则除非调用stop永不退出
void setLifeTime(int seconds) { lifetime_ = seconds; }
void setAutoReconnect(bool b) { autoReconnect_ = b; } void setAutoReconnect(bool b) { autoReconnect_ = b; }
bool start(const std::string& ip, uint16_t port, std::string& msg);
// start_in_thread 是否开启工作线程。
// 设置为 true 则开启工作线程,可以跨线程调用 stop 主动停止
// 设置为 false 则阻塞调用,不能调用 stop,如果设置了生命周期长度,将在到期后自动退出,否则永不退出
bool start(const std::string& ip, uint16_t port, std::string& msg, bool start_in_thread = true);
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_; }
...@@ -72,6 +78,7 @@ protected: ...@@ -72,6 +78,7 @@ protected:
OnTimerCallback onTimer_ = nullptr; OnTimerCallback onTimer_ = nullptr;
int timeout_ = 5; int timeout_ = 5;
bool strictTimer_ = false; bool strictTimer_ = false;
int lifetime_ = 0;
std::mutex mutex_ = {}; std::mutex mutex_ = {};
std::chrono::steady_clock::time_point lastTimeSendData = {}; std::chrono::steady_clock::time_point lastTimeSendData = {};
......
...@@ -160,6 +160,13 @@ void simple_libevent_clients::BaseClient::set_auto_reconnect(bool b) ...@@ -160,6 +160,13 @@ void simple_libevent_clients::BaseClient::set_auto_reconnect(bool b)
struct simple_libevent_clients::PrivateImpl struct simple_libevent_clients::PrivateImpl
{ {
struct WorkerThreadContext;
struct ReconnectContext {
WorkerThreadContext* context = nullptr;
BaseClient* client = nullptr;
};
struct WorkerThreadContext { struct WorkerThreadContext {
simple_libevent_clients* ctx = nullptr; simple_libevent_clients* ctx = nullptr;
int thread_id = 0; int thread_id = 0;
...@@ -201,6 +208,8 @@ struct simple_libevent_clients::PrivateImpl ...@@ -201,6 +208,8 @@ struct simple_libevent_clients::PrivateImpl
client->privateData->bev = bev; client->privateData->bev = bev;
client->privateData->thread_id = thread_id; client->privateData->thread_id = thread_id;
client->privateData->client_id = client_id_to_connect++; client->privateData->client_id = client_id_to_connect++;
client->privateData->server_ip = ip;
client->privateData->server_port = port;
bufferevent_setcb(bev, readcb, writecb, eventcb, this); bufferevent_setcb(bev, readcb, writecb, eventcb, this);
bufferevent_enable(bev, EV_READ | EV_WRITE); bufferevent_enable(bev, EV_READ | EV_WRITE);
...@@ -306,14 +315,29 @@ struct simple_libevent_clients::PrivateImpl ...@@ -306,14 +315,29 @@ struct simple_libevent_clients::PrivateImpl
if (context->ctx->onConn_) { if (context->ctx->onConn_) {
context->ctx->onConn_(up, msg, client, context->ctx->userData_); context->ctx->onConn_(up, msg, client, context->ctx->userData_);
} }
if (!up) { if (!up) {
if (client->privateData->timer) { if (client->privateData->timer) {
event_del(client->privateData->timer); event_del(client->privateData->timer);
client->privateData->timer = nullptr; client->privateData->timer = nullptr;
} }
std::lock_guard<std::mutex> lg(context->mutex);
context->clients.erase(fd); if (client->privateData->lifetimer) {
delete client; event_del(client->privateData->lifetimer);
client->privateData->lifetimer = nullptr;
}
{
std::lock_guard<std::mutex> lg(context->mutex);
context->clients.erase(fd);
}
if (client->privateData->auto_reconnect) {
struct timeval tv = { 3, 0 };
event_add(event_new(context->base, -1, 0, reconn_timercb, new ReconnectContext{ context, client }), &tv);
} else {
delete client;
}
} }
} }
...@@ -321,6 +345,51 @@ struct simple_libevent_clients::PrivateImpl ...@@ -321,6 +345,51 @@ struct simple_libevent_clients::PrivateImpl
bufferevent_free(bev); bufferevent_free(bev);
} }
} }
static void reconn_timercb(evutil_socket_t, short, void* user_data)
{
ReconnectContext* rctx = (ReconnectContext*)user_data;
BaseClient* client = rctx->client;
bool ok = false;
std::string msg;
do {
msg = "Reconnecting to " + client->server_ip() + ":" + std::to_string(client->server_port());
auto bev = bufferevent_socket_new(rctx->context->base, -1, BEV_OPT_CLOSE_ON_FREE);
if (!bev) {
msg += (" allocate bufferevent failed");
break;
}
client->privateData->bev = bev;
bufferevent_setcb(bev, readcb, writecb, eventcb, rctx->context);
bufferevent_enable(bev, EV_READ | EV_WRITE);
sockaddr_in addr = { 0 };
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = inet_addr(client->server_ip().data());
addr.sin_port = htons(client->server_port());
if (bufferevent_socket_connect(bev, (const sockaddr*)(&addr), sizeof(addr)) < 0) {
client->privateData->fd = (int)bufferevent_getfd(bev);
int err = evutil_socket_geterror(client->privateData->fd);
msg += " error starting connection: " + std::to_string(err) + evutil_socket_error_to_string(err);
} else {
ok = true;
client->privateData->fd = (int)bufferevent_getfd(bev);
std::lock_guard<std::mutex> lg(rctx->context->mutex);
rctx->context->clients[client->privateData->fd] = client;
}
} while (0);
if (rctx->context->ctx->onConn_) {
rctx->context->ctx->onConn_(false, msg, client, rctx->context->ctx->userData_);
}
if (!ok) {
delete client;
}
delete rctx;
}
}; };
typedef WorkerThreadContext* WorkerThreadContextPtr; typedef WorkerThreadContext* WorkerThreadContextPtr;
std::vector<WorkerThreadContextPtr> contexts{}; std::vector<WorkerThreadContextPtr> contexts{};
...@@ -432,6 +501,14 @@ void simple_libevent_clients::BaseClient::set_timer(OnTimerCallback cb, void* us ...@@ -432,6 +501,14 @@ void simple_libevent_clients::BaseClient::set_timer(OnTimerCallback cb, void* us
void simple_libevent_clients::BaseClient::set_lifetime(int seconds) void simple_libevent_clients::BaseClient::set_lifetime(int seconds)
{ {
if (seconds < 0) {
if (privateData->lifetimer) {
event_del(privateData->lifetimer);
privateData->lifetimer = nullptr;
}
return;
}
privateData->lifetime = seconds; privateData->lifetime = seconds;
if (privateData->lifetimer) { if (privateData->lifetimer) {
event_del(privateData->lifetimer); event_del(privateData->lifetimer);
......
...@@ -70,6 +70,7 @@ public: ...@@ -70,6 +70,7 @@ public:
void updateLastTimeComm(); void updateLastTimeComm();
void set_auto_reconnect(bool b); void set_auto_reconnect(bool b);
void set_timer(OnTimerCallback cb, void* user_data, int seconds); void set_timer(OnTimerCallback cb, void* user_data, int seconds);
// set to -1 for live until peer disconnected
void set_lifetime(int seconds); void set_lifetime(int seconds);
struct PrivateData; struct PrivateData;
......
...@@ -119,6 +119,7 @@ ...@@ -119,6 +119,7 @@
<PrecompiledHeader>NotUsing</PrecompiledHeader> <PrecompiledHeader>NotUsing</PrecompiledHeader>
<PrecompiledHeaderFile>pch.h</PrecompiledHeaderFile> <PrecompiledHeaderFile>pch.h</PrecompiledHeaderFile>
<AdditionalIncludeDirectories>$(DEVLIBS)\libevent-2.1.12-stable-install\include;%(AdditionalIncludeDirectories)</AdditionalIncludeDirectories> <AdditionalIncludeDirectories>$(DEVLIBS)\libevent-2.1.12-stable-install\include;%(AdditionalIncludeDirectories)</AdditionalIncludeDirectories>
<RuntimeLibrary>MultiThreaded</RuntimeLibrary>
</ClCompile> </ClCompile>
<Link> <Link>
<SubSystem> <SubSystem>
......
#include "../../jlib/net/Client.h" #include "../../jlib/net/simple_libevent_client.h"
#include "../../jlib/net/Server.h" #include "../../jlib/net/simple_libevent_server.h"
#include "../../jlib/net/simple_libevent_clients.h"
using namespace jlib::net; using namespace jlib::net;
int main() int main()
{ {
server::Server server; simple_libevent_server server;
client::Client client; simple_libevent_client client;
simple_libevent_clients clients(nullptr, nullptr, nullptr, nullptr, 1, nullptr);
} }
...@@ -94,7 +94,7 @@ ...@@ -94,7 +94,7 @@
<SubSystem>Console</SubSystem> <SubSystem>Console</SubSystem>
<GenerateDebugInformation>true</GenerateDebugInformation> <GenerateDebugInformation>true</GenerateDebugInformation>
<AdditionalLibraryDirectories>$(SolutionDir)$(Configuration)\;%(AdditionalLibraryDirectories)</AdditionalLibraryDirectories> <AdditionalLibraryDirectories>$(SolutionDir)$(Configuration)\;%(AdditionalLibraryDirectories)</AdditionalLibraryDirectories>
<AdditionalDependencies>simple_libevent_client.lib;simple_libevent_server.lib;%(AdditionalDependencies)</AdditionalDependencies> <AdditionalDependencies>simple_libevent_client.lib;simple_libevent_server.lib;simple_libevent_clients.lib;%(AdditionalDependencies)</AdditionalDependencies>
</Link> </Link>
</ItemDefinitionGroup> </ItemDefinitionGroup>
<ItemDefinitionGroup Condition="'$(Configuration)|$(Platform)'=='Release|Win32'"> <ItemDefinitionGroup Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">
......
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