xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

enh: refact client server messages

This commit is contained in:
dapan1121 2022-11-18 11:43:03 +08:00
parent 1b0c98433e
commit 3f88dbe256
4 changed files with 66 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/common/tmsg.h
View File

@ -1741,6 +1741,9 @@ typedef struct {
int32_t tSerializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *pReq); int32_t tSerializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *pReq);
int32_t tDeserializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *pReq); int32_t tDeserializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *pReq);
int32_t tSerializeSQueryTableRsp(void *buf, int32_t bufLen, SQueryTableRsp *pRsp);
int32_t tDeserializeSQueryTableRsp(void *buf, int32_t bufLen, SQueryTableRsp *pRsp);
typedef struct { typedef struct {
int32_t code; int32_t code;

2
source/client/src/clientTmq.c
View File

@ -756,7 +756,7 @@ void tmqSendHbReq(void* param, void* tmrId) {
} }
sendInfo->msgInfo = (SDataBuf){ sendInfo->msgInfo = (SDataBuf){
.pData = pReq, .pData = pReq,
.len = sizeof(SMqHbReq), .len = tlen,
.handle = NULL, .handle = NULL,
}; };

37
source/common/src/tmsg.c
View File

@ -4778,6 +4778,43 @@ int32_t tDeserializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *pReq)
return 0; return 0;
} }
int32_t tSerializeSQueryTableRsp(void *buf, int32_t bufLen, SQueryTableRsp *pRsp) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->code) < 0) return -1;
if (tEncodeCStr(&encoder, pRsp->tbFName) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->sversion) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->tversion) < 0) return -1;
if (tEncodeI64(&encoder, pRsp->affectedRows) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSQueryTableRsp(void *buf, int32_t bufLen, SQueryTableRsp *pRsp) {
SDecoder decoder = {0};
tDecoderInit(&decoder, (char *)buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->code) < 0) return -1;
if (tDecodeCStrTo(&decoder, pRsp->tbFName) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->sversion) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->tversion) < 0) return -1;
if (tDecodeI64(&decoder, &pRsp->affectedRows) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
int32_t tSerializeSSchedulerHbReq(void *buf, int32_t bufLen, SSchedulerHbReq *pReq) { int32_t tSerializeSSchedulerHbReq(void *buf, int32_t bufLen, SSchedulerHbReq *pReq) {
int32_t headLen = sizeof(SMsgHead); int32_t headLen = sizeof(SMsgHead);

32
source/libs/qworker/src/qwMsg.c
View File

@ -65,19 +65,37 @@ int32_t qwBuildAndSendErrorRsp(int32_t rspType, SRpcHandleInfo *pConn, int32_t c
int32_t qwBuildAndSendQueryRsp(int32_t rspType, SRpcHandleInfo *pConn, int32_t code, SQWTaskCtx *ctx) { int32_t qwBuildAndSendQueryRsp(int32_t rspType, SRpcHandleInfo *pConn, int32_t code, SQWTaskCtx *ctx) {
STbVerInfo *tbInfo = ctx ? &ctx->tbInfo : NULL; STbVerInfo *tbInfo = ctx ? &ctx->tbInfo : NULL;
int64_t affectedRows = ctx ? ctx->affectedRows : 0; int64_t affectedRows = ctx ? ctx->affectedRows : 0;
SQueryTableRsp *pRsp = (SQueryTableRsp *)rpcMallocCont(sizeof(SQueryTableRsp)); SQueryTableRsp rsp = {0};
pRsp->code = htonl(code); rsp.code = code;
pRsp->affectedRows = htobe64(affectedRows); rsp.affectedRows = affectedRows;
if (tbInfo) { if (tbInfo) {
strcpy(pRsp->tbFName, tbInfo->tbFName); strcpy(rsp.tbFName, tbInfo->tbFName);
pRsp->sversion = htonl(tbInfo->sversion); rsp.sversion = tbInfo->sversion;
pRsp->tversion = htonl(tbInfo->tversion); rsp.tversion = tbInfo->tversion;
}
int32_t msgSize = tSerializeSQueryTableRsp(NULL, 0, &rsp);
if (msgSize < 0) {
qError("tSerializeSQueryTableRsp failed");
QW_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
void *pRsp = rpcMallocCont(msgSize);
if (NULL == pRsp) {
qError("rpcMallocCont %d failed", msgSize);
QW_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
if (tSerializeSQueryTableRsp(pRsp, msgSize, &rsp) < 0) {
qError("tSerializeSQueryTableRsp %d failed", msgSize);
QW_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
} }
SRpcMsg rpcRsp = { SRpcMsg rpcRsp = {
.msgType = rspType, .msgType = rspType,
.pCont = pRsp, .pCont = pRsp,
.contLen = sizeof(*pRsp), .contLen = msgSize,
.code = code, .code = code,
.info = *pConn, .info = *pConn,
}; };