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

handle except

This commit is contained in:
yihaoDeng 2022-03-20 18:58:00 +08:00
parent 843477e3a1
commit e657b37e41
1 changed files with 21 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -168,34 +168,20 @@ static void uvHandleActivityTimeout(uv_timer_t* handle) {
} }
static void uvHandleReq(SSrvConn* pConn) { static void uvHandleReq(SSrvConn* pConn) {
SRecvInfo info;
SRecvInfo* p = &info;
SConnBuffer* pBuf = &pConn->readBuf; SConnBuffer* pBuf = &pConn->readBuf;
p->msg = pBuf->buf; char* msg = pBuf->buf;
p->msgLen = pBuf->len; uint32_t msgLen = pBuf->len;
p->ip = 0;
p->port = 0;
p->shandle = pConn->pTransInst; //
p->thandle = pConn;
p->chandle = NULL;
STransMsgHead* pHead = (STransMsgHead*)p->msg; STransMsgHead* pHead = (STransMsgHead*)msg;
if (pHead->secured == 1) { if (pHead->secured == 1) {
STransUserMsg* uMsg = (STransUserMsg*)((char*)p->msg + p->msgLen - sizeof(STransUserMsg)); STransUserMsg* uMsg = (STransUserMsg*)((char*)msg + msgLen - sizeof(STransUserMsg));
memcpy(pConn->user, uMsg->user, tListLen(uMsg->user)); memcpy(pConn->user, uMsg->user, tListLen(uMsg->user));
memcpy(pConn->secret, uMsg->secret, tListLen(uMsg->secret)); memcpy(pConn->secret, uMsg->secret, tListLen(uMsg->secret));
} }
pHead->code = htonl(pHead->code); pHead->code = htonl(pHead->code);
pHead->msgLen = htonl(pHead->msgLen);
int32_t dlen = 0; if (pHead->secured == 1) {
if (transDecompressMsg(NULL, 0, NULL)) { pHead->msgLen -= sizeof(STransUserMsg);
// add compress later
// pHead = rpcDecompresSTransMsg(pHead);
} else {
pHead->msgLen = htonl(pHead->msgLen);
if (pHead->secured == 1) {
pHead->msgLen -= sizeof(STransUserMsg);
}
} }
CONN_SHOULD_RELEASE(pConn, pHead); CONN_SHOULD_RELEASE(pConn, pHead);
@ -232,7 +218,7 @@ static void uvHandleReq(SSrvConn* pConn) {
transMsg.handle = pConn; transMsg.handle = pConn;
} }
STrans* pTransInst = (STrans*)p->shandle; STrans* pTransInst = pConn->hostThrd;
(*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL); (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
// uv_timer_start(&pConn->pTimer, uvHandleActivityTimeout, pRpc->idleTime * 10000, 0); // uv_timer_start(&pConn->pTimer, uvHandleActivityTimeout, pRpc->idleTime * 10000, 0);
} }
@ -341,21 +327,27 @@ static void uvPrepareSendData(SSrvMsg* smsg, uv_buf_t* wb) {
} }
STransMsgHead* pHead = transHeadFromCont(pMsg->pCont); STransMsgHead* pHead = transHeadFromCont(pMsg->pCont);
pHead->secured = pMsg->code == 0 ? 1 : 0; // // pHead->secured = pMsg->code == 0 ? 1 : 0; //
pHead->msgType = smsg->pConn->inType + 1; if (!pConn->secured) {
pConn->secured = pMsg->code == 0 ? 1 : 0;
}
pHead->secured = pConn->secured;
if (pConn->status == ConnNormal) {
pHead->msgType = pConn->inType + 1;
} else {
pHead->msgType = smsg->type == Release ? 0 : pMsg->msgType;
}
pHead->release = smsg->type == Release ? 1 : 0; pHead->release = smsg->type == Release ? 1 : 0;
pHead->code = htonl(pMsg->code); pHead->code = htonl(pMsg->code);
// add more info
char* msg = (char*)pHead; char* msg = (char*)pHead;
int32_t len = transMsgLenFromCont(pMsg->contLen); int32_t len = transMsgLenFromCont(pMsg->contLen);
if (transCompressMsg(msg, len, NULL)) {
// impl later
}
tDebug("server conn %p %s is sent to %s:%d, local info: %s:%d", pConn, TMSG_INFO(pHead->msgType), tDebug("server conn %p %s is sent to %s:%d, local info: %s:%d", pConn, TMSG_INFO(pHead->msgType),
taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->locaddr.sin_addr), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->locaddr.sin_addr),
ntohs(pConn->locaddr.sin_port)); ntohs(pConn->locaddr.sin_port));
pHead->msgLen = htonl(len); pHead->msgLen = htonl(len);
wb->base = msg; wb->base = msg;
wb->len = len; wb->len = len;
} }