From 3278a2309bea41993a12f0e7906fa474d41c2f6b Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 21 Apr 2022 17:51:33 +0800 Subject: [PATCH] enh(rpc):add auth --- include/common/tmsg.h | 8 ++++++++ source/common/src/tmsg.c | 22 ++++++++++++++++++++++ source/libs/transport/src/trans.c | 9 ++++----- source/libs/transport/src/transCli.c | 11 ++--------- source/libs/transport/src/transSrv.c | 5 ----- 5 files changed, 36 insertions(+), 19 deletions(-) diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 809390a9d2..b9a561d790 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -326,6 +326,13 @@ int32_t tDecodeSEpSet(SCoder* pDecoder, SEpSet* pEp); int32_t taosEncodeSEpSet(void** buf, const SEpSet* pEp); void* taosDecodeSEpSet(const void* buf, SEpSet* pEp); +typedef struct { + SEpSet epSet; +} SMEpSet; + +int32_t tSerializeSMEpSet(void* buf, int32_t bufLen, SMEpSet* pReq); +int32_t tDeserializeSMEpSet(void* buf, int32_t buflen, SMEpSet* pReq); + typedef struct { int8_t connType; int32_t pid; @@ -2690,6 +2697,7 @@ static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* p } return buf; } + #pragma pack(pop) #ifdef __cplusplus diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 3622390f45..e0ae67597e 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -828,6 +828,28 @@ void tFreeSMAltertbReq(SMAltertbReq *pReq) { taosArrayDestroy(pReq->pFields); pReq->pFields = NULL; } +int32_t tSerializeSMEpSet(void *buf, int32_t bufLen, SMEpSet *pReq) { + SCoder encoder = {0}; + tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER); + if (tEncodeSEpSet(&encoder, &pReq->epSet) < 0) { + return -1; + } + tEndEncode(&encoder); + int32_t tlen = encoder.pos; + tCoderClear(&encoder); + return tlen; +} +int32_t tDeserializeSMEpSet(void *buf, int32_t bufLen, SMEpSet *pReq) { + SCoder decoder = {0}; + tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER); + if (tDecodeSEpSet(&decoder, &pReq->epSet) < 0) { + return -1; + } + + tEndDecode(&decoder); + tCoderClear(&decoder); + return 0; +} int32_t tSerializeSMCreateSmaReq(void *buf, int32_t bufLen, SMCreateSmaReq *pReq) { SCoder encoder = {0}; diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c index ebb90338cd..5f9af2eb48 100644 --- a/source/libs/transport/src/trans.c +++ b/source/libs/transport/src/trans.c @@ -100,11 +100,10 @@ void rpcSendRedirectRsp(void* thandle, const SEpSet* pEpSet) { SRpcMsg rpcMsg; memset(&rpcMsg, 0, sizeof(rpcMsg)); - rpcMsg.contLen = sizeof(SEpSet); - rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen); - if (rpcMsg.pCont == NULL) return; - - memcpy(rpcMsg.pCont, pEpSet, sizeof(SEpSet)); + SMEpSet msg = {.epSet = *pEpSet}; + int32_t len = tSerializeSMEpSet(NULL, 0, &msg); + rpcMsg.pCont = rpcMallocCont(len); + tSerializeSMEpSet(rpcMsg.pCont, len, &msg); rpcMsg.code = TSDB_CODE_RPC_REDIRECT; rpcMsg.handle = thandle; diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index b81310b90b..b2d0e7f020 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -31,12 +31,8 @@ typedef struct SCliConn { int hThrdIdx; STransCtx ctx; - bool broken; // link broken or not - ConnStatus status; // - int release; // 1: release - // spi configure - char spi; - char secured; + bool broken; // link broken or not + ConnStatus status; // char* ip; uint32_t port; @@ -44,7 +40,6 @@ typedef struct SCliConn { // debug and log info struct sockaddr_in addr; struct sockaddr_in locaddr; - } SCliConn; typedef struct SCliMsg { @@ -303,8 +298,6 @@ void cliHandleResp(SCliConn* conn) { TMSG_INFO(pHead->msgType), taosInetNtoa(conn->addr.sin_addr), ntohs(conn->addr.sin_port), taosInetNtoa(conn->locaddr.sin_addr), ntohs(conn->locaddr.sin_port), transMsg.contLen); - conn->secured = pHead->secured; - if (pCtx == NULL && CONN_NO_PERSIST_BY_APP(conn)) { tTrace("except, server continue send while cli ignore it"); // transUnrefCliHandle(conn); diff --git a/source/libs/transport/src/transSrv.c b/source/libs/transport/src/transSrv.c index ec66f3e8df..b1b2781859 100644 --- a/source/libs/transport/src/transSrv.c +++ b/source/libs/transport/src/transSrv.c @@ -30,7 +30,6 @@ typedef struct SSrvConn { uv_timer_t pTimer; queue queue; - int ref; int persist; // persist connection or not SConnBuffer readBuf; // read buf, int inType; @@ -692,8 +691,6 @@ static void uvDestroyConn(uv_handle_t* handle) { if (thrd->quit && QUEUE_IS_EMPTY(&thrd->conn)) { tTrace("work thread quit"); uv_walk(thrd->loop, uvWalkCb, NULL); - // uv_loop_close(thrd->loop); - // uv_stop(thrd->loop); } } @@ -756,8 +753,6 @@ void uvHandleQuit(SSrvMsg* msg, SWorkThrdObj* thrd) { thrd->quit = true; if (QUEUE_IS_EMPTY(&thrd->conn)) { uv_walk(thrd->loop, uvWalkCb, NULL); - // uv_loop_close(thrd->loop); - // uv_stop(thrd->loop); } else { destroyAllConn(thrd); }