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

refactor code

This commit is contained in:
Yihao Deng 2024-07-11 10:14:46 +00:00
parent 4e3a785574
commit 0a138a5078
1 changed files with 17 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
source/libs/transport/src/transCli.c
View File

@ -470,8 +470,11 @@ void cliHandleResp_shareConn(SCliConn* conn) {
transMsg.info.ahandle = pCtx ? pCtx->ahandle : NULL; transMsg.info.ahandle = pCtx ? pCtx->ahandle : NULL;
STraceId* trace = &transMsg.info.traceId; STraceId* trace = &transMsg.info.traceId;
if (cliAppCb(conn, &transMsg, pMsg) != 0) { int32_t ret = cliAppCb(conn, &transMsg, pMsg);
if (ret != 0) {
return; return;
} else {
destroyCmsg(pMsg);
} }
} }
void cliHandleResp(SCliConn* conn) { void cliHandleResp(SCliConn* conn) {
@ -1131,12 +1134,13 @@ static void cliHandleBatch_shareConnExcept(SCliConn* conn) {
int32_t code = -1; int32_t code = -1;
SCliThrd* pThrd = conn->hostThrd; SCliThrd* pThrd = conn->hostThrd;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
for (int i = 0; i < transQueueSize(&conn->cliMsgs); i++) { while (!transQueueEmpty(&conn->cliMsgs)) {
SCliMsg* pMsg = transQueueGet(&conn->cliMsgs, i); SCliMsg* pMsg = transQueuePop(&conn->cliMsgs);
ASSERT(pMsg->type != Release); ASSERT(pMsg->type != Release);
ASSERT(REQUEST_NO_RESP(&pMsg->msg) == 0); ASSERT(REQUEST_NO_RESP(&pMsg->msg) == 0);
STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL; STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL;
STransMsg transMsg = {0}; STransMsg transMsg = {0};
transMsg.code = code == -1 ? (conn->broken ? TSDB_CODE_RPC_BROKEN_LINK : TSDB_CODE_RPC_NETWORK_UNAVAIL) : code; transMsg.code = code == -1 ? (conn->broken ? TSDB_CODE_RPC_BROKEN_LINK : TSDB_CODE_RPC_NETWORK_UNAVAIL) : code;
transMsg.msgType = pMsg ? pMsg->msg.msgType + 1 : 0; transMsg.msgType = pMsg ? pMsg->msg.msgType + 1 : 0;
@ -1144,20 +1148,18 @@ static void cliHandleBatch_shareConnExcept(SCliConn* conn) {
transMsg.info.cliVer = pTransInst->compatibilityVer; transMsg.info.cliVer = pTransInst->compatibilityVer;
transMsg.info.ahandle = pCtx->ahandle; transMsg.info.ahandle = pCtx->ahandle;
int32_t ret = cliAppCb(conn, &transMsg, pMsg); pMsg->seqNum = 0;
if (ret != 0) { code = cliAppCb(conn, &transMsg, pMsg);
return; if (code != 0) {
continue;
} else {
// already notify user
destroyCmsg(pMsg);
} }
} }
// SCliConn* conn = req->data; if (T_REF_VAL_GET(conn) > 1) transUnrefCliHandle(conn);
// if (status != 0) { transUnrefCliHandle(conn);
// tDebug("%s conn %p failed to send batch msg, reason:%s", CONN_GET_INST_LABEL(conn), conn, uv_err_name(status));
// return;
// }
// uv_read_start((uv_stream_t*)conn->stream, cliAllocRecvBufferCb, cliRecvCb);
// taosMemoryFree(req);
} }
static void cliSendBatch_shareConnCb(uv_write_t* req, int status) { static void cliSendBatch_shareConnCb(uv_write_t* req, int status) {
SCliConn* conn = req->data; SCliConn* conn = req->data;