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

handle close except

This commit is contained in:
yihaoDeng 2022-01-24 23:42:41 +08:00
parent b34a5bdf51
commit 2fd97f4c7d
4 changed files with 33 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
source/libs/transport/inc/transComm.h
View File

@ -182,7 +182,7 @@ typedef struct {
#define TRANS_RESERVE_SIZE (sizeof(STranConnCtx))
#define TRANS_MSG_OVERHEAD (sizeof(STransMsgHead) + sizeof(STransDigestMsg))
#define TRANS_MSG_OVERHEAD (sizeof(STransMsgHead))
#define transHeadFromCont(cont) ((STransMsgHead*)((char*)cont - sizeof(STransMsgHead)))
#define transContFromHead(msg) (msg + sizeof(STransMsgHead))
#define transMsgLenFromCont(contLen) (contLen + sizeof(STransMsgHead))
@ -201,6 +201,7 @@ bool transDecompressMsg(char* msg, int32_t len, int32_t* flen);
void transConnCtxDestroy(STransConnCtx* ctx);
void transFreeMsg(void* msg);
typedef struct SConnBuffer {
char* buf;
int len;

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

@ -107,8 +107,17 @@ static void clientHandleResp(SCliConn* conn) {
SRpcInfo* pRpc = pCtx->pTransInst;
SRpcMsg rpcMsg;
rpcMsg.pCont = conn->readBuf.buf;
rpcMsg.contLen = conn->readBuf.len;
STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf);
pHead->code = htonl(pHead->code);
pHead->msgLen = htonl(pHead->msgLen);
rpcMsg.contLen = transContLenFromMsg(pHead->msgLen);
rpcMsg.pCont = transContFromHead(pHead);
rpcMsg.code = pHead->code;
rpcMsg.msgType = pHead->msgType;
// rpcMsg.pCont = conn->readBuf.buf;
// rpcMsg.contLen = conn->readBuf.len;
rpcMsg.ahandle = pCtx->ahandle;
(pRpc->cfp)(NULL, &rpcMsg, NULL);
conn->notifyCount += 1;
@ -128,8 +137,12 @@ static void clientHandleExcept(SCliConn* pConn) {
STransConnCtx* pCtx = pMsg->ctx;
SRpcInfo* pRpc = pCtx->pTransInst;
SRpcMsg rpcMsg;
transFreeMsg((pMsg->msg.pCont));
pMsg->msg.pCont = NULL;
SRpcMsg rpcMsg = {0};
rpcMsg.ahandle = pCtx->ahandle;
rpcMsg.code = -1;
// SRpcInfo* pRpc = pMsg->ctx->pRpc;
(pRpc->cfp)(NULL, &rpcMsg, NULL);
@ -306,6 +319,10 @@ static void clientDestroy(uv_handle_t* handle) {
static void clientWriteCb(uv_write_t* req, int status) {
SCliConn* pConn = req->data;
SCliMsg* pMsg = pConn->data;
transFreeMsg((pMsg->msg.pCont));
pMsg->msg.pCont = NULL;
if (status == 0) {
tDebug("conn %p data already was written out", pConn);
} else {

7
source/libs/transport/src/transComm.c
View File

@ -191,4 +191,11 @@ void transConnCtxDestroy(STransConnCtx* ctx) {
free(ctx->ip);
free(ctx);
}
void transFreeMsg(void* msg) {
if (msg == NULL) {
return;
}
free((char*)msg - sizeof(STransMsgHead));
}
#endif

4
source/libs/transport/src/transSrv.c
View File

@ -321,6 +321,10 @@ void uvOnWriteCb(uv_write_t* req, int status) {
buf->len = 0;
memset(buf->buf, 0, buf->cap);
buf->left = -1;
SRpcMsg* pMsg = &conn->sendMsg;
transFreeMsg(pMsg->pCont);
if (status == 0) {
tDebug("conn %p data already was written on stream", conn);
} else {