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

fix compile

This commit is contained in:
Liu Jicong 2021-10-27 17:53:47 +08:00
parent 5e14c85609
commit 35869f2ed3
2 changed files with 32 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
source/server/vnode/src/vnodeReadMsg.c
View File

@ -225,16 +225,16 @@ int32_t vnodeProcessConsumeMsg(SVnode *pVnode, SReadMsg *pRead) {
tmqMsgHead msgHead = pConsumeMsg->head; tmqMsgHead msgHead = pConsumeMsg->head;
//extract head //extract head
STQ *pTq = pVnode->pTQ; STQ *pTq = pVnode->pTQ;
tqGroupHandle *pHandle = tqFindGHandleBycId(pTq, msgHead.clientId); /*tqBufferHandle *pHandle = tqGetHandle(pTq, msgHead.clientId);*/
//return msg if offset not moved //return msg if offset not moved
if(pConsumeMsg->commitOffset == pHandle->consumeOffset) { /*if(pConsumeMsg->commitOffset == pHandle->consumeOffset) {*/
//return msg //return msg
return 0; /*return 0;*/
} /*}*/
//or move offset //or move offset
tqMoveOffsetToNext(pHandle); /*tqMoveOffsetToNext(pHandle);*/
//fetch or register context //fetch or register context
tqFetchMsg(pHandle, pRead); /*tqFetchMsg(pHandle, pRead);*/
//judge mode, tail read or catch up read //judge mode, tail read or catch up read
/*int64_t lastVer = walLastVer(pVnode->wal);*/ /*int64_t lastVer = walLastVer(pVnode->wal);*/
//launch new query //launch new query

26
source/server/vnode/tq/src/tq.c
View File

@ -148,6 +148,19 @@ static int tqFetch(tqGroupHandle* ghandle, void** msg) {
return totSize; return totSize;
} }
tqGroupHandle* tqGetGroupHandle(STQ* pTq, int64_t cId) {
return NULL;
}
int tqLaunchQuery(tqGroupHandle* ghandle) {
return 0;
}
int tqSendLaunchQuery(STQ* pTq, int64_t topicId, int64_t cgId, void* query) {
return 0;
}
/*int tqMoveOffsetToNext(tqGroupHandle* ghandle) {*/ /*int tqMoveOffsetToNext(tqGroupHandle* ghandle) {*/
/*return 0;*/ /*return 0;*/
/*}*/ /*}*/
@ -268,3 +281,16 @@ int tqDeserializeBufItem(const void* pBytes, tqBufferItem **pBufItem) {
return 0; return 0;
} }
int tqGetGHandleSSize(const tqGroupHandle *gHandle) {
return 0;
}
int tqListHandleSSize(const tqListHandle *listHandle) {
return 0;
}
int tqBufHandleSSize(const tqBufferHandle *bufHandle) {
return 0;
}
int tqBufItemSSize(const tqBufferItem *bufItem) {
return 0;
}