Germany/hunting
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Backup:TcpModule.

Browse Source
This commit is contained in:
Fancy code 2024-04-11 15:57:49 +08:00
parent 6f6e8eb7cd
commit a7fc63285c
18 changed files with 238 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
middleware/McuManager/src/McuDevice.cpp
View File

@ -79,8 +79,8 @@ const StatusCode McuDevice::UnInit(void)
DeleteAllAsk();
return CreateStatusCode(STATUS_CODE_OK);
}
size_t McuDevice::WriteData(const void *buff, const size_t buffLength, std::shared_ptr<VProtocolContext> &context,
const unsigned int &serialNumber)
ssize_t McuDevice::WriteData(const void *buff, const size_t buffLength, std::shared_ptr<VProtocolContext> &context,
const unsigned int &serialNumber)
{
constexpr size_t WRITE_ERROR = -1;
size_t length = WRITE_ERROR;

4
middleware/McuManager/src/McuDevice.h
View File

@ -29,8 +29,8 @@ public:
const StatusCode UnInit(void) override;
public:
size_t WriteData(const void *buff, const size_t buffLength, std::shared_ptr<VProtocolContext> &context,
const unsigned int &serialNumber) override;
ssize_t WriteData(const void *buff, const size_t buffLength, std::shared_ptr<VProtocolContext> &context,
const unsigned int &serialNumber) override;
public:
void GetIpcMissionReply(const unsigned int &serialNumber, const unsigned char &mission) override;

25
test/utils/TcpModule/src/TcpModule_Test.cpp
View File

@ -19,33 +19,42 @@
#include <thread>
namespace TcpModuleTest
{
// ../output_files/test/bin/TcpModuleTest --gtest_filter=TcpModuleTest.UNIT_TcpModule_AUTO_IllegalObject
// ../output_files/test/bin/TcpModuleTest --gtest_filter=TcpModuleTest.UNIT_TcpModule_EXAMPLE_AUTO_IllegalObject
/**
* TcpModule module api will not crash when object is illegal.
*/
TEST(TcpModuleTest, UNIT_TcpModule_AUTO_IllegalObject)
TEST(TcpModuleTest, UNIT_TcpModule_EXAMPLE_AUTO_IllegalObject)
{
static void *tcpClientAccept = nullptr;
TcpParam param = {
TcpServerParam tcpServerparam = {
.mIp = "127.0.0.1",
.mPort = 9876,
.mReadFunc = [](const void *p, const size_t len, void *context) -> void {
return;
},
.mAcceptClientFunc = [](void *object, const char *ip) -> void {
LogInfo("accept client, peer ip: %s", ip);
tcpClientAccept = object;
return;
},
};
TcpClientParam param = {
.mIp = "127.0.0.1",
.mPort = 9876,
.mReadFunc = [](const void *data, const ssize_t len, const void *context) -> void {
LogInfo("read data: %s", (char *)data);
return;
},
};
CreateLogModule();
ILogInit(LOG_INSTANCE_TYPE_END);
void *tcpServer = CreateTcpServer(param);
void *tcpServer = CreateTcpServer(tcpServerparam);
std::this_thread::sleep_for(std::chrono::milliseconds(500));
void *tcpClient = CreateTcpClient(param);
std::this_thread::sleep_for(std::chrono::milliseconds(100));
TcpClientWrite(tcpClient, "123456789", strlen("123456789"));
std::this_thread::sleep_for(std::chrono::milliseconds(3000));
std::this_thread::sleep_for(std::chrono::milliseconds(100));
if (nullptr != tcpClientAccept) {
AcceptClientWrite(tcpClientAccept, "9876543210", strlen("9876543210"));
}
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
if (nullptr != tcpClient) {
FreeTcpClient(tcpClient);
}

4
utils/McuProtocol/include/McuProtocol.h
View File

@ -68,8 +68,8 @@ protected:
{
return std::make_shared<VProtocolBase>();
}
virtual size_t WriteData(const void *buff, const size_t buffLength, std::shared_ptr<VProtocolContext> &context,
const unsigned int &serialNumber)
virtual ssize_t WriteData(const void *buff, const size_t buffLength, std::shared_ptr<VProtocolContext> &context,
const unsigned int &serialNumber)
{
return 0;
}

29
utils/TcpModule/include/TcpModule.h
View File

@ -16,23 +16,36 @@
#define TCP_MODULE_H
#include <stddef.h>
#include <sys/types.h>
#define TCP_MODULE_WRITE_ERROR -1;
#ifdef __cplusplus
extern "C" {
#endif
typedef void (*TcpReadFunction)(const void *, const size_t, void *);
typedef void (*TcpReadFunction)(const void *, const ssize_t, const void *);
typedef void (*TcpAcceptClientFunction)(void *, const char *);
typedef void (*SocketClosedFunction)(const void *);
typedef struct tcp_server_parm
{
const char *mIp;
const int mPort;
TcpAcceptClientFunction mAcceptClientFunc;
} TcpServerParam;
typedef struct tcp_parm
{
const char *mIp;
const int mPort;
TcpReadFunction mReadFunc; // TODO: delete this memeber
TcpAcceptClientFunction mAcceptClientFunc;
} TcpParam;
void *CreateTcpServer(const TcpParam param);
TcpReadFunction mReadFunc;
SocketClosedFunction mClosedFunc;
} TcpClientParam;
typedef struct client_accept_parm
{
TcpReadFunction mReadFunc;
SocketClosedFunction mClosedFunc;
} ClientAcceptParam;
void *CreateTcpServer(const TcpServerParam param);
void FreeTcpServer(void *object);
int AcceptClientSetReadFunc(void *object, TcpReadFunction readFunc);
int AcceptClientWrite(void *object, void *buf, const size_t bufLenght);
void *CreateTcpClient(const TcpParam param);
void AcceptClientSetParam(void *object, const ClientAcceptParam param);
ssize_t AcceptClientWrite(void *object, const void *buf, const size_t bufLenght);
void *CreateTcpClient(const TcpClientParam param);
void FreeTcpClient(void *object);
ssize_t TcpClientWrite(void *object, const void *buf, const size_t bufLenght);
#ifdef __cplusplus

9
utils/TcpModule/src/ITcpClient.cpp
View File

@ -14,6 +14,7 @@
*/
#include "ITcpClient.h"
#include "ILog.h"
#include "TcpModule.h"
#include <cstring>
void ITcpClient::Init(void)
{
@ -21,9 +22,15 @@ void ITcpClient::Init(void)
void ITcpClient::UnInit(void)
{
}
void ITcpClient::Readed(const void *data, size_t length)
{
}
ssize_t ITcpClient::Write(const void *buf, const size_t bufLenght)
{
return WRITE_ERROR;
return TCP_MODULE_WRITE_ERROR;
}
void ITcpClient::Closed(void)
{
}
static const char *TCP_CLIENT_NAME = "tcp_client";
const char *GetTcpClientModuleName(void)

3
utils/TcpModule/src/ITcpClient.h
View File

@ -16,7 +16,6 @@
#define I_TCP_CLIENT_H
#include "StatusCode.h"
#include <memory>
constexpr int WRITE_ERROR = -1;
class ITcpClient
{
public:
@ -24,7 +23,9 @@ public:
virtual ~ITcpClient() = default;
virtual void Init(void);
virtual void UnInit(void);
virtual void Readed(const void *data, size_t length);
virtual ssize_t Write(const void *buf, const size_t bufLenght);
virtual void Closed(void);
};
typedef struct i_tcp_client_header
{

23
utils/TcpModule/src/ITcpServer.cpp
View File

@ -14,10 +14,21 @@
*/
#include "ITcpServer.h"
#include "ILog.h"
#include "TcpModule.h"
#include <cstring>
void ITcpClientAccept::SetParam(const ClientAcceptParam &param)
{
}
void ITcpClientAccept::Readed(const void *data, size_t length)
{
}
ssize_t ITcpClientAccept::Write(const void *data, size_t length)
{
return TCP_MODULE_WRITE_ERROR;
}
void ITcpClientAccept::Closed(void)
{
}
void ITcpServer::Init(void)
{
}
@ -33,4 +44,16 @@ const char *GetTcpServerModuleName(void)
const char *GetTcpClientAcceptName(void)
{
return TCP_CLIENT_ACCEPT_NAME;
}
bool TcpClientAcceptObjectCheck(void *object)
{
if (nullptr == object) {
LogError("nullptr object!\n");
return false;
}
if (*((const char **)(((char *)object) - sizeof(ITcpServerHeader))) != GetTcpClientAcceptName()) {
LogError("Illegal object!\n");
return false;
}
return true;
}

5
utils/TcpModule/src/ITcpServer.h
View File

@ -15,13 +15,17 @@
#ifndef I_TCP_SERVER_H
#define I_TCP_SERVER_H
#include "StatusCode.h"
#include "TcpModule.h"
#include <memory>
class ITcpClientAccept
{
public:
ITcpClientAccept() = default;
virtual ~ITcpClientAccept() = default;
virtual void SetParam(const ClientAcceptParam &param);
virtual void Readed(const void *data, size_t length);
virtual ssize_t Write(const void *data, size_t length);
virtual void Closed(void);
};
class ITcpServer
{
@ -47,4 +51,5 @@ typedef struct tcp_client_accept
} TcpClientAccept;
const char *GetTcpServerModuleName(void);
const char *GetTcpClientAcceptName(void);
bool TcpClientAcceptObjectCheck(void *object);
#endif

52
utils/TcpModule/src/TcpClientImpl.cpp
View File

@ -15,38 +15,26 @@
#include "TcpClientImpl.h"
#include "ILog.h"
#include <cstring>
TcpClientImpl::TcpClientImpl(const TcpParam param) : mParam(param)
{
}
static void on_message(hio_t *io, void *buf, int len)
{
LogInfo("onmessage: %.*s\n", len, (char *)buf);
TcpClientImpl *cli = (TcpClientImpl *)hevent_userdata(io);
// ...
TcpClientImpl *tcpClient = (TcpClientImpl *)hevent_userdata(io);
tcpClient->Readed(buf, len);
}
static void on_connect(hio_t *io)
{
LogInfo("onconnect: connfd=%d\n", hio_fd(io));
TcpClientImpl *cli = (TcpClientImpl *)hevent_userdata(io);
// cli->connected = 1;
// hio_write(io, "hello\r\n", 7);
hio_setcb_read(io, on_message);
hio_read(io);
}
static void on_close(hio_t *io)
{
LogInfo("onclose: connfd=%d error=%d\n", hio_fd(io), hio_error(io));
TcpClientImpl *cli = (TcpClientImpl *)hevent_userdata(io);
// cli->connected = 0;
// reconnect
// if (cli->reconn_setting && reconn_setting_can_retry(cli->reconn_setting)) {
// uint32_t delay = reconn_setting_calc_delay(cli->reconn_setting);
// LogInfo("reconnect cnt=%d, delay=%d ...\n", cli->reconn_setting->cur_retry_cnt,
// cli->reconn_setting->cur_delay); cli->reconn_timer = htimer_add(cli->mLoop, reconnect_timer_cb, delay, 1);
// hevent_set_userdata(cli->reconn_timer, cli);
// }
TcpClientImpl *tcpClient = (TcpClientImpl *)hevent_userdata(io);
tcpClient->Closed();
}
TcpClientImpl::TcpClientImpl(const TcpClientParam &param, const void *object) : mParam(param), mObjectThis(object)
{
}
void TcpClientImpl::Init(void)
{
@ -85,14 +73,30 @@ void TcpClientImpl::UnInit(void)
mLoop = nullptr;
mIo = nullptr;
}
void TcpClientImpl::Readed(const void *data, size_t length)
{
if (nullptr != mParam.mReadFunc) {
mParam.mReadFunc(data, length, nullptr);
return;
}
LogError("mParam.mReadFunc is nullptr.\n");
}
ssize_t TcpClientImpl::Write(const void *buf, const size_t bufLenght)
{
if (nullptr == mIo) {
LogError("mIo is nullptr.\n");
return WRITE_ERROR;
return TCP_MODULE_WRITE_ERROR;
}
return hio_write(mIo, buf, bufLenght);
}
void TcpClientImpl::Closed(void)
{
if (nullptr != mParam.mClosedFunc) {
mParam.mClosedFunc(mObjectThis);
return;
}
LogWarning("mParam.mClosedFunc is null\n");
}
void TcpClientImpl::Loop(void)
{
if (nullptr == mLoop) {
@ -101,7 +105,7 @@ void TcpClientImpl::Loop(void)
}
hloop_run(mLoop);
}
std::shared_ptr<ITcpClient> *NewTcpClient(const TcpParam param)
std::shared_ptr<ITcpClient> *NewTcpClient(const TcpClientParam &param)
{
LogInfo("Create tcp server object.\n");
TcpClient *impl = (TcpClient *)malloc(sizeof(TcpClient));
@ -112,6 +116,8 @@ std::shared_ptr<ITcpClient> *NewTcpClient(const TcpParam param)
TcpClient tmp;
memcpy((void *)impl, (void *)&tmp, sizeof(TcpClient));
impl->mHeader.mCheckName = GetTcpClientModuleName();
impl->mTcpClient = std::make_shared<TcpClientImpl>(param);
return (std::shared_ptr<ITcpClient> *)(((char *)impl) + sizeof(ITcpClientHeader));
std::shared_ptr<ITcpClient> *objectThis =
(std::shared_ptr<ITcpClient> *)(((char *)impl) + sizeof(ITcpClientHeader));
impl->mTcpClient = std::make_shared<TcpClientImpl>(param, objectThis);
return objectThis;
}

11
utils/TcpModule/src/TcpClientImpl.h
View File

@ -24,18 +24,21 @@
class TcpClientImpl : public ITcpClient, public std::enable_shared_from_this<TcpClientImpl>
{
public:
TcpClientImpl(const TcpParam param);
TcpClientImpl(const TcpClientParam &param, const void *object);
virtual ~TcpClientImpl() = default;
void Init(void) override;
void UnInit(void) override;
void Readed(const void *data, size_t length) override;
ssize_t Write(const void *buf, const size_t bufLenght) override;
void Closed(void) override;
void Loop(void);
private:
hloop_t* mLoop;
hloop_t *mLoop;
hio_t *mIo;
const TcpParam mParam;
const TcpClientParam mParam;
std::thread mTcpClientThread;
const void *mObjectThis;
};
std::shared_ptr<ITcpClient> *NewTcpClient(const TcpParam param);
std::shared_ptr<ITcpClient> *NewTcpClient(const TcpClientParam &param);
#endif

17
utils/TcpModule/src/TcpModule.cpp
View File

@ -42,7 +42,7 @@ static bool TcpClientObjectCheck(void *object)
}
return true;
}
void *CreateTcpServer(const TcpParam param)
void *CreateTcpServer(const TcpServerParam param)
{
std::shared_ptr<ITcpServer> *server = TcpModuleMakePtr::GetInstance()->CreateTcpServer(param);
if (nullptr != *server) {
@ -58,7 +58,20 @@ void FreeTcpServer(void *object)
free(((char *)object) - sizeof(ITcpServerHeader)); // TODO: bug?
}
}
void *CreateTcpClient(const TcpParam param)
void AcceptClientSetParam(void *object, const ClientAcceptParam param)
{
if (TcpClientAcceptObjectCheck(object) == true) {
(*(std::shared_ptr<ITcpClientAccept> *)object)->SetParam(param);
}
}
ssize_t AcceptClientWrite(void *object, const void *buf, const size_t bufLenght)
{
if (TcpClientAcceptObjectCheck(object) == true) {
return (*(std::shared_ptr<ITcpClientAccept> *)object)->Write(buf, bufLenght);
}
return TCP_MODULE_WRITE_ERROR;
}
void *CreateTcpClient(const TcpClientParam param)
{
std::shared_ptr<ITcpClient> *client = TcpModuleMakePtr::GetInstance()->CreateTcpClient(param);
if (nullptr != *client) {

4
utils/TcpModule/src/TcpModuleImpl.cpp
View File

@ -20,7 +20,7 @@
// {
// return TCP_MODULE_NAME;
// }
// void *NewTcpModuleImpl(const TcpParam &tcpParam)
// void *NewTcpModuleImpl(const TcpClientParam &tcpParam)
// {
// if (nullptr == tcpParam.mIp) {
// LogError("Parament error, nullptr == tcpParam.mIp\n");
@ -34,7 +34,7 @@
// impl->mTcpImpl = std::make_shared<TcpModuleImpl>(tcpParam);
// return (void *)(((char *)impl) + sizeof(TcpModuleHeader));
// }
// void *NewTcpServer(const TcpParam &tcpParam)
// void *NewTcpServer(const TcpClientParam &tcpParam)
// {
// return nullptr;
// }

8
utils/TcpModule/src/TcpModuleImpl.h
View File

@ -25,14 +25,14 @@
// class TcpModuleImpl
// {
// public:
// TcpModuleImpl(const TcpParam &uatrInfo);
// TcpModuleImpl(const TcpClientParam &uatrInfo);
// virtual ~TcpModuleImpl() = default;
// private:
// const StatusCode SetConfig(void);
// private:
// const TcpParam mUatrInfo;
// const TcpClientParam mUatrInfo;
// int mFd;
// };
// TODO: There may be a CPU byte alignment bug.
@ -42,8 +42,8 @@
// TcpModuleHeader mHeader;
// std::shared_ptr<TcpModuleImpl> mTcpImpl;
// } TcpServer;
// void *NewTcpModuleImpl(const TcpParam &tcpParam);
// void *NewTcpServer(const TcpParam &tcpParam);
// void *NewTcpModuleImpl(const TcpClientParam &tcpParam);
// void *NewTcpServer(const TcpClientParam &tcpParam);
// static inline TcpServer *TcpModuleImplConvert(void *object)
// {
// return ((TcpServer *)(((char *)object) - sizeof(TcpModuleHeader)));

6
utils/TcpModule/src/TcpModuleMakePtr.cpp
View File

@ -14,8 +14,8 @@
*/
#include "TcpModuleMakePtr.h"
#include "ILog.h"
#include "TcpServerImpl.h"
#include "TcpClientImpl.h"
#include "TcpServerImpl.h"
std::shared_ptr<TcpModuleMakePtr> &TcpModuleMakePtr::GetInstance(std::shared_ptr<TcpModuleMakePtr> *impl)
{
static auto instance = std::make_shared<TcpModuleMakePtr>();
@ -24,11 +24,11 @@ std::shared_ptr<TcpModuleMakePtr> &TcpModuleMakePtr::GetInstance(std::shared_ptr
}
return instance;
}
std::shared_ptr<ITcpServer> *TcpModuleMakePtr::CreateTcpServer(const TcpParam param)
std::shared_ptr<ITcpServer> *TcpModuleMakePtr::CreateTcpServer(const TcpServerParam &param)
{
return NewTcpServer(param);
}
std::shared_ptr<ITcpClient> *TcpModuleMakePtr::CreateTcpClient(const TcpParam param)
std::shared_ptr<ITcpClient> *TcpModuleMakePtr::CreateTcpClient(const TcpClientParam &param)
{
return NewTcpClient(param);
}

8
utils/TcpModule/src/TcpModuleMakePtr.h
View File

@ -14,10 +14,10 @@
*/
#ifndef TCP_MODULE_MAKE_PTR_H
#define TCP_MODULE_MAKE_PTR_H
#include "ITcpClient.h"
#include "ITcpServer.h"
#include "StatusCode.h"
#include "TcpModule.h"
#include "ITcpServer.h"
#include "ITcpClient.h"
#include <memory>
class TcpModuleMakePtr
{
@ -25,7 +25,7 @@ public:
TcpModuleMakePtr() = default;
virtual ~TcpModuleMakePtr() = default;
static std::shared_ptr<TcpModuleMakePtr> &GetInstance(std::shared_ptr<TcpModuleMakePtr> *impl = nullptr);
virtual std::shared_ptr<ITcpServer> *CreateTcpServer(const TcpParam param);
virtual std::shared_ptr<ITcpClient> *CreateTcpClient(const TcpParam param);
virtual std::shared_ptr<ITcpServer> *CreateTcpServer(const TcpServerParam &param);
virtual std::shared_ptr<ITcpClient> *CreateTcpClient(const TcpClientParam &param);
};
#endif // !TCP_MODULE_MAKE_PTR_H

159
utils/TcpModule/src/TcpServerImpl.cpp
View File

@ -15,28 +15,11 @@
#include "TcpServerImpl.h"
#include "ILog.h"
#include "TcpServerHandle.h"
static bool TcpClientAcceptObjectCheck(void *object)
{
if (nullptr == object) {
LogError("nullptr object!\n");
return false;
}
if (*((const char **)(((char *)object) - sizeof(ITcpServerHeader))) != GetTcpClientAcceptName()) {
LogError("Illegal object!\n");
return false;
}
return true;
}
static void on_close(hio_t *io)
{
LogInfo("on_close fd=%d error=%d\n", hio_fd(io), hio_error(io));
// std::shared_ptr<ITcpServer> server;
// if (TcpServerHandle::GetInstance()->GetServer(hio_fd(io), server) == true) {
// std::shared_ptr<TcpServerImpl> impl = std::dynamic_pointer_cast<TcpServerImpl>(server);
// if (impl != nullptr) {
// impl->ClosedEvent();
// }
// }
TcpServerImpl *server = (TcpServerImpl *)hevent_userdata(io);
server->RemoveClient(io);
}
static void on_recv(hio_t *io, void *buf, int readbytes)
{
@ -46,19 +29,9 @@ static void on_recv(hio_t *io, void *buf, int readbytes)
LogInfo(
"[%s] <=> [%s]\n", SOCKADDR_STR(hio_localaddr(io), localaddrstr), SOCKADDR_STR(hio_peeraddr(io), peeraddrstr));
LogInfo("< %.*s", readbytes, (char *)buf);
ITcpClientAccept *client = (ITcpClientAccept *)hevent_userdata(io);
client->Readed((const char *)buf, readbytes);
// server->RemoveClient(hio_fd(io));
// std::shared_ptr<ITcpServer> server;
// if (TcpServerHandle::GetInstance()->GetServer(hio_fd(io), server) == true) {
// std::shared_ptr<TcpServerImpl> impl = std::dynamic_pointer_cast<TcpServerImpl>(server);
// if (impl != nullptr) {
// impl->TcpServerReaded(buf, readbytes);
// }
// }
// echo
// LogInfo("> %.*s", readbytes, (char *)buf);
// hio_write(io, buf, readbytes);
TcpServerImpl *server = (TcpServerImpl *)hevent_userdata(io);
std::shared_ptr<ITcpClientAccept> *client = server->GetClient(io);
(*client)->Readed((const char *)buf, readbytes);
}
static void on_accept(hio_t *io)
{
@ -74,25 +47,47 @@ static void on_accept(hio_t *io)
hio_setcb_read(io, on_recv);
std::shared_ptr<ITcpClientAccept> *client = NewTcpClientAccept(io);
TcpServerImpl *server = (TcpServerImpl *)hevent_userdata(io);
// std::shared_ptr<ITcpServer> server;
// if (TcpServerHandle::GetInstance()->GetServer(hio_fd(io), server) == true) {
// std::shared_ptr<TcpServerImpl> impl = std::dynamic_pointer_cast<TcpServerImpl>(server);
// if (impl != nullptr) {
// impl->SetIo(io);
// }
// }
server->AddClient(io, client);
hevent_set_userdata(io, (*client).get());
hio_read_start(io);
}
TcpClientAcceptImpl::TcpClientAcceptImpl(const hio_t *io) : mIo(io)
TcpClientAcceptImpl::TcpClientAcceptImpl(const hio_t *io, const void *object) : mIo(io), mObjectThis(object)
{
mParam.mReadFunc = nullptr;
mParam.mClosedFunc = nullptr;
}
void TcpClientAcceptImpl::SetParam(const ClientAcceptParam &param)
{
std::lock_guard<std::mutex> locker(mMutex);
mParam = param;
}
void TcpClientAcceptImpl::Readed(const void *data, size_t length)
{
LogInfo("TcpClientAcceptImpl::Readed\n");
std::lock_guard<std::mutex> locker(mMutex);
if (nullptr != mParam.mReadFunc) {
mParam.mReadFunc(data, length, mObjectThis);
return;
}
LogWarning("mParam.mClosedFunc is null\n");
}
TcpServerImpl::TcpServerImpl(const TcpParam param) : mParam(param)
ssize_t TcpClientAcceptImpl::Write(const void *data, size_t length)
{
if (mIo) {
hio_t *io = (hio_t *)mIo;
return hio_write(io, data, length);
}
LogError("mIo is null\n");
return TCP_MODULE_WRITE_ERROR;
}
void TcpClientAcceptImpl::Closed(void)
{
std::lock_guard<std::mutex> locker(mMutex);
if (nullptr != mParam.mClosedFunc) {
mParam.mClosedFunc(mObjectThis);
return;
}
LogWarning("mParam.mClosedFunc is null\n");
}
TcpServerImpl::TcpServerImpl(const TcpServerParam param) : mParam(param)
{
mLoop = nullptr;
mIo = nullptr;
@ -114,7 +109,6 @@ void TcpServerImpl::Init(void)
hevent_set_userdata(listenio, this);
mIo = listenio;
std::shared_ptr<ITcpServer> server = shared_from_this();
// TcpServerHandle::GetInstance()->AddServer(hio_fd(listenio), server);
std::shared_ptr<TcpServerImpl> impl = std::dynamic_pointer_cast<TcpServerImpl>(server);
auto recvThread = [](std::shared_ptr<TcpServerImpl> tcpServer) {
tcpServer->Loop();
@ -136,22 +130,6 @@ void TcpServerImpl::UnInit(void)
mLoop = nullptr;
mIo = nullptr;
}
void TcpServerImpl::TcpServerReaded(const void *buf, size_t length)
{
}
void TcpServerImpl::TcpServerWrite(const void *buf, size_t length)
{
if (mIo) {
hio_write(mIo, buf, length);
return;
}
LogError("mIo is null\n");
}
void TcpServerImpl::ClosedEvent(void)
{
mLoop = nullptr;
mIo = nullptr;
}
void TcpServerImpl::Loop(void)
{
if (nullptr == mLoop) {
@ -162,6 +140,7 @@ void TcpServerImpl::Loop(void)
}
void TcpServerImpl::AddClient(hio_t *io, std::shared_ptr<ITcpClientAccept> *client)
{
std::lock_guard<std::mutex> locker(mMutex);
char localaddrstr[SOCKADDR_STRLEN] = {0};
char peeraddrstr[SOCKADDR_STRLEN] = {0};
LogInfo("accept connfd=%d [%s] <= [%s]\n",
@ -173,37 +152,51 @@ void TcpServerImpl::AddClient(hio_t *io, std::shared_ptr<ITcpClientAccept> *clie
mParam.mAcceptClientFunc(client, peeraddrstr);
}
}
std::shared_ptr<ITcpClientAccept> *TcpServerImpl::GetClient(hio_t *io)
{
std::lock_guard<std::mutex> locker(mMutex);
auto it = mClients.find(hio_fd(io));
if (it != mClients.end()) {
return it->second;
}
else {
LogError("GetClient failed, client not exit.\n");
return nullptr;
}
}
void TcpServerImpl::RemoveClient(hio_t *io)
{
std::lock_guard<std::mutex> locker(mMutex);
auto it = mClients.find(hio_fd(io));
if (it != mClients.end()) {
void *object = (void *)it->second;
if (TcpClientAcceptObjectCheck(object) == true) {
(*(std::shared_ptr<ITcpClientAccept> *)object)->Closed();
(*(std::shared_ptr<ITcpClientAccept> *)object).reset();
free(((char *)object) - sizeof(ITcpServerHeader));
}
mClients.erase(it);
}
else {
LogError("RemoveClient failed, client not exit.\n");
}
}
void TcpServerImpl::FreeClients(void)
{
std::lock_guard<std::mutex> locker(mMutex);
for (auto &client : mClients) {
if (nullptr != client.second) {
// delete client.second;
// client.second = nullptr;
void *object = (void *)client.second;
if (TcpClientAcceptObjectCheck(object) == true) {
// (*(std::shared_ptr<ITcpClientAccept> *)object)->UnInit();
(*(std::shared_ptr<ITcpClientAccept> *)object)->Closed();
(*(std::shared_ptr<ITcpClientAccept> *)object).reset();
free(((char *)object) - sizeof(ITcpServerHeader)); // TODO: bug?
free(((char *)object) - sizeof(ITcpServerHeader));
}
}
}
// for (auto it = mClients.begin(); it != mClients.end();) {
// // 检查指针是否为空
// if (it->second != nullptr) {
// // 释放 shared_ptr，这将减少引用计数
// it->second->reset();
// // 删除指针
// delete it->second;
// // 从 map 中删除条目
// it = mClients.erase(it);
// }
// else {
// // 指针为空，继续迭代
// ++it;
// }
// }
mClients.clear();
}
std::shared_ptr<ITcpServer> *NewTcpServer(const TcpParam param)
std::shared_ptr<ITcpServer> *NewTcpServer(const TcpServerParam &param)
{
LogInfo("Create tcp server object.\n");
TcpServer *impl = (TcpServer *)malloc(sizeof(TcpServer));
@ -228,6 +221,8 @@ std::shared_ptr<ITcpClientAccept> *NewTcpClientAccept(const hio_t *io)
TcpClientAccept tmp;
memcpy((void *)impl, (void *)&tmp, sizeof(TcpClientAccept));
impl->mHeader.mCheckName = GetTcpClientAcceptName();
impl->mTcpClientAccept = std::make_shared<TcpClientAcceptImpl>(io);
return (std::shared_ptr<ITcpClientAccept> *)(((char *)impl) + sizeof(ITcpServerHeader));
std::shared_ptr<ITcpClientAccept> *objectThis =
(std::shared_ptr<ITcpClientAccept> *)(((char *)impl) + sizeof(ITcpServerHeader));
impl->mTcpClientAccept = std::make_shared<TcpClientAcceptImpl>(io, objectThis);
return objectThis;
}

23
utils/TcpModule/src/TcpServerImpl.h
View File

@ -20,38 +20,45 @@
#include "hsocket.h"
#include "hssl.h"
#include <map>
#include <mutex>
#include <thread>
class TcpClientAcceptImpl : public ITcpClientAccept
class TcpClientAcceptImpl : public ITcpClientAccept, public std::enable_shared_from_this<TcpClientAcceptImpl>
{
public:
TcpClientAcceptImpl(const hio_t *io);
TcpClientAcceptImpl(const hio_t *io, const void *object);
virtual ~TcpClientAcceptImpl() = default;
void SetParam(const ClientAcceptParam &param) override;
void Readed(const void *data, size_t length) override;
ssize_t Write(const void *data, size_t length) override;
void Closed(void) override;
private:
std::mutex mMutex;
const hio_t *mIo;
ClientAcceptParam mParam;
const void *mObjectThis;
};
class TcpServerImpl : public ITcpServer, public std::enable_shared_from_this<TcpServerImpl>
{
public:
TcpServerImpl(const TcpParam param);
TcpServerImpl(const TcpServerParam param);
virtual ~TcpServerImpl() = default;
void Init(void) override;
void UnInit(void) override;
void TcpServerReaded(const void *buf, size_t length);
void TcpServerWrite(const void *buf, size_t length);
void ClosedEvent(void);
void Loop(void);
void AddClient(hio_t *io, std::shared_ptr<ITcpClientAccept> *client);
std::shared_ptr<ITcpClientAccept> *GetClient(hio_t *io);
void RemoveClient(hio_t *io);
void FreeClients(void);
private:
std::mutex mMutex;
hloop_t *mLoop;
hio_t *mIo;
const TcpParam mParam;
const TcpServerParam mParam;
std::thread mTcpServerThread;
std::map<int, std::shared_ptr<ITcpClientAccept> *> mClients;
};
std::shared_ptr<ITcpServer> *NewTcpServer(const TcpParam param);
std::shared_ptr<ITcpServer> *NewTcpServer(const TcpServerParam &param);
std::shared_ptr<ITcpClientAccept> *NewTcpClientAccept(const hio_t *io);
#endif