From 668109924576c1f345ea8f06d79bdcb58bdaa460 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Sat, 12 Nov 2022 21:31:01 +0800 Subject: [PATCH] refact: build sync heartbeat reply --- source/libs/sync/inc/syncMessage.h | 22 +-- source/libs/sync/src/syncMain.c | 28 ++- source/libs/sync/src/syncMessage.c | 162 ++---------------- .../sync/test/sync_test_lib/inc/syncTest.h | 18 ++ .../test/sync_test_lib/src/syncMessageDebug.c | 156 ++++++++++++++++- 5 files changed, 199 insertions(+), 187 deletions(-) diff --git a/source/libs/sync/inc/syncMessage.h b/source/libs/sync/inc/syncMessage.h index b9bca4e616..4d66b37137 100644 --- a/source/libs/sync/inc/syncMessage.h +++ b/source/libs/sync/inc/syncMessage.h @@ -140,25 +140,6 @@ typedef struct SyncHeartbeatReply { int64_t startTime; } SyncHeartbeatReply; -SyncHeartbeatReply* syncHeartbeatReplyBuild(int32_t vgId); -void syncHeartbeatReplyDestroy(SyncHeartbeatReply* pMsg); -void syncHeartbeatReplySerialize(const SyncHeartbeatReply* pMsg, char* buf, uint32_t bufLen); -void syncHeartbeatReplyDeserialize(const char* buf, uint32_t len, SyncHeartbeatReply* pMsg); -char* syncHeartbeatReplySerialize2(const SyncHeartbeatReply* pMsg, uint32_t* len); -SyncHeartbeatReply* syncHeartbeatReplyDeserialize2(const char* buf, uint32_t len); -void syncHeartbeatReply2RpcMsg(const SyncHeartbeatReply* pMsg, SRpcMsg* pRpcMsg); -void syncHeartbeatReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncHeartbeatReply* pMsg); -SyncHeartbeatReply* syncHeartbeatReplyFromRpcMsg2(const SRpcMsg* pRpcMsg); -cJSON* syncHeartbeatReply2Json(const SyncHeartbeatReply* pMsg); -char* syncHeartbeatReply2Str(const SyncHeartbeatReply* pMsg); - -// for debug ---------------------- -void syncHeartbeatReplyPrint(const SyncHeartbeatReply* pMsg); -void syncHeartbeatReplyPrint2(char* s, const SyncHeartbeatReply* pMsg); -void syncHeartbeatReplyLog(const SyncHeartbeatReply* pMsg); -void syncHeartbeatReplyLog2(char* s, const SyncHeartbeatReply* pMsg); - -// --------------------------------------------- typedef struct SyncPreSnapshot { uint32_t bytes; int32_t vgId; @@ -401,7 +382,7 @@ int32_t syncNodeOnSnapshot(SSyncNode* ths, SyncSnapshotSend* pMsg); int32_t syncNodeOnSnapshotReply(SSyncNode* ths, SyncSnapshotRsp* pMsg); int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pMsg); -int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, SyncHeartbeatReply* pMsg); +int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, const SRpcMsg* pMsg); int32_t syncNodeOnClientRequest(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIndex); int32_t syncNodeOnLocalCmd(SSyncNode* ths, SyncLocalCmd* pMsg); @@ -422,6 +403,7 @@ int32_t syncBuildRequestVoteReply(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildAppendEntries(SRpcMsg* pMsg, int32_t dataLen, int32_t vgId); int32_t syncBuildAppendEntriesReply(SRpcMsg* pMsg, int32_t vgId); int32_t syncBuildHeartbeat(SRpcMsg* pMsg, int32_t vgId); +int32_t syncBuildHeartbeatReply(SRpcMsg* pMsg, int32_t vgId); #ifdef __cplusplus } diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 88e5fde1b9..bd7b119105 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -136,21 +136,19 @@ int32_t syncProcessMsg(int64_t rid, SRpcMsg* pMsg) { if (pMsg->msgType == TDMT_SYNC_HEARTBEAT) { code = syncNodeOnHeartbeat(pSyncNode, pMsg); } else if (pMsg->msgType == TDMT_SYNC_HEARTBEAT_REPLY) { - SyncHeartbeatReply* pSyncMsg = syncHeartbeatReplyFromRpcMsg2(pMsg); - code = syncNodeOnHeartbeatReply(pSyncNode, pSyncMsg); - syncHeartbeatReplyDestroy(pSyncMsg); + code = syncNodeOnHeartbeatReply(pSyncNode, pMsg); } else if (pMsg->msgType == TDMT_SYNC_TIMEOUT) { code = syncNodeOnTimer(pSyncNode, pMsg); } else if (pMsg->msgType == TDMT_SYNC_CLIENT_REQUEST) { code = syncNodeOnClientRequest(pSyncNode, pMsg, NULL); } else if (pMsg->msgType == TDMT_SYNC_REQUEST_VOTE) { - syncNodeOnRequestVote(pSyncNode, pMsg); + code = syncNodeOnRequestVote(pSyncNode, pMsg); } else if (pMsg->msgType == TDMT_SYNC_REQUEST_VOTE_REPLY) { code = syncNodeOnRequestVoteReply(pSyncNode, pMsg); } else if (pMsg->msgType == TDMT_SYNC_APPEND_ENTRIES) { - syncNodeOnAppendEntries(pSyncNode, pMsg); + code = syncNodeOnAppendEntries(pSyncNode, pMsg); } else if (pMsg->msgType == TDMT_SYNC_APPEND_ENTRIES_REPLY) { - syncNodeOnAppendEntriesReply(pSyncNode, pMsg); + code = syncNodeOnAppendEntriesReply(pSyncNode, pMsg); } else if (pMsg->msgType == TDMT_SYNC_SNAPSHOT_SEND) { SyncSnapshotSend* pSyncMsg = syncSnapshotSendFromRpcMsg2(pMsg); code = syncNodeOnSnapshot(pSyncNode, pSyncMsg); @@ -1912,7 +1910,7 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) { if (pSyncNode->replicaNum > 1) { if (timerLogicClock == msgLogicClock) { - SRpcMsg rpcMsg = {0}; + SRpcMsg rpcMsg = {0}; (void)syncBuildHeartbeat(&rpcMsg, pSyncNode->vgId); SyncHeartbeat* pSyncMsg = rpcMsg.pCont; @@ -2010,15 +2008,15 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) { SyncHeartbeat* pMsg = pRpcMsg->pCont; syncLogRecvHeartbeat(ths, pMsg, ""); - SyncHeartbeatReply* pMsgReply = syncHeartbeatReplyBuild(ths->vgId); + SRpcMsg rpcMsg = {0}; + (void)syncBuildHeartbeatReply(&rpcMsg, ths->vgId); + + SyncHeartbeatReply* pMsgReply = rpcMsg.pCont; pMsgReply->destId = pMsg->srcId; pMsgReply->srcId = ths->myRaftId; pMsgReply->term = ths->pRaftStore->currentTerm; pMsgReply->privateTerm = 8864; // magic number - SRpcMsg rpcMsg; - syncHeartbeatReply2RpcMsg(pMsgReply, &rpcMsg); - if (pMsg->term == ths->pRaftStore->currentTerm && ths->state != TAOS_SYNC_STATE_LEADER) { syncNodeResetElectTimer(ths); ths->minMatchIndex = pMsg->minMatchIndex; @@ -2075,17 +2073,15 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) { // reply syncNodeSendMsgById(&pMsgReply->destId, ths, &rpcMsg); - syncHeartbeatReplyDestroy(pMsgReply); - return 0; } -int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, SyncHeartbeatReply* pMsg) { +int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) { + SyncHeartbeatReply* pMsg = pRpcMsg->pCont; syncLogRecvHeartbeatReply(ths, pMsg, ""); // update last reply time, make decision whether the other node is alive or not - syncIndexMgrSetRecvTime(ths->pMatchIndex, &(pMsg->destId), pMsg->startTime); - + syncIndexMgrSetRecvTime(ths->pMatchIndex, &pMsg->destId, pMsg->startTime); return 0; } diff --git a/source/libs/sync/src/syncMessage.c b/source/libs/sync/src/syncMessage.c index 6941e83478..cc68d370a8 100644 --- a/source/libs/sync/src/syncMessage.c +++ b/source/libs/sync/src/syncMessage.c @@ -172,159 +172,21 @@ int32_t syncBuildHeartbeat(SRpcMsg* pMsg, int32_t vgId) { return 0; } - -// ---- message process SyncHeartbeatReply---- -SyncHeartbeatReply* syncHeartbeatReplyBuild(int32_t vgId) { - uint32_t bytes = sizeof(SyncHeartbeatReply); - SyncHeartbeatReply* pMsg = taosMemoryMalloc(bytes); - memset(pMsg, 0, bytes); - pMsg->bytes = bytes; - pMsg->vgId = vgId; +int32_t syncBuildHeartbeatReply(SRpcMsg* pMsg, int32_t vgId) { + int32_t bytes = sizeof(SyncHeartbeatReply); + pMsg->pCont = rpcMallocCont(bytes); pMsg->msgType = TDMT_SYNC_HEARTBEAT_REPLY; - return pMsg; -} - -void syncHeartbeatReplyDestroy(SyncHeartbeatReply* pMsg) { - if (pMsg != NULL) { - taosMemoryFree(pMsg); - } -} - -void syncHeartbeatReplySerialize(const SyncHeartbeatReply* pMsg, char* buf, uint32_t bufLen) { - ASSERT(pMsg->bytes <= bufLen); - memcpy(buf, pMsg, pMsg->bytes); -} - -void syncHeartbeatReplyDeserialize(const char* buf, uint32_t len, SyncHeartbeatReply* pMsg) { - memcpy(pMsg, buf, len); - ASSERT(len == pMsg->bytes); -} - -char* syncHeartbeatReplySerialize2(const SyncHeartbeatReply* pMsg, uint32_t* len) { - char* buf = taosMemoryMalloc(pMsg->bytes); - ASSERT(buf != NULL); - syncHeartbeatReplySerialize(pMsg, buf, pMsg->bytes); - if (len != NULL) { - *len = pMsg->bytes; - } - return buf; -} - -SyncHeartbeatReply* syncHeartbeatReplyDeserialize2(const char* buf, uint32_t len) { - uint32_t bytes = *((uint32_t*)buf); - SyncHeartbeatReply* pMsg = taosMemoryMalloc(bytes); - ASSERT(pMsg != NULL); - syncHeartbeatReplyDeserialize(buf, len, pMsg); - ASSERT(len == pMsg->bytes); - return pMsg; -} - -void syncHeartbeatReply2RpcMsg(const SyncHeartbeatReply* pMsg, SRpcMsg* pRpcMsg) { - memset(pRpcMsg, 0, sizeof(*pRpcMsg)); - pRpcMsg->msgType = pMsg->msgType; - pRpcMsg->contLen = pMsg->bytes; - pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen); - syncHeartbeatReplySerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen); -} - -void syncHeartbeatReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncHeartbeatReply* pMsg) { - syncHeartbeatReplyDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg); -} - -SyncHeartbeatReply* syncHeartbeatReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) { - SyncHeartbeatReply* pMsg = syncHeartbeatReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); - ASSERT(pMsg != NULL); - return pMsg; -} - -cJSON* syncHeartbeatReply2Json(const SyncHeartbeatReply* pMsg) { - char u64buf[128] = {0}; - cJSON* pRoot = cJSON_CreateObject(); - - if (pMsg != NULL) { - cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes); - cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId); - cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType); - - cJSON* pSrcId = cJSON_CreateObject(); - snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->srcId.addr); - cJSON_AddStringToObject(pSrcId, "addr", u64buf); - { - uint64_t u64 = pMsg->srcId.addr; - cJSON* pTmp = pSrcId; - char host[128] = {0}; - uint16_t port; - syncUtilU642Addr(u64, host, sizeof(host), &port); - cJSON_AddStringToObject(pTmp, "addr_host", host); - cJSON_AddNumberToObject(pTmp, "addr_port", port); - } - cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId); - cJSON_AddItemToObject(pRoot, "srcId", pSrcId); - - cJSON* pDestId = cJSON_CreateObject(); - snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->destId.addr); - cJSON_AddStringToObject(pDestId, "addr", u64buf); - { - uint64_t u64 = pMsg->destId.addr; - cJSON* pTmp = pDestId; - char host[128] = {0}; - uint16_t port; - syncUtilU642Addr(u64, host, sizeof(host), &port); - cJSON_AddStringToObject(pTmp, "addr_host", host); - cJSON_AddNumberToObject(pTmp, "addr_port", port); - } - cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId); - cJSON_AddItemToObject(pRoot, "destId", pDestId); - - snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->privateTerm); - cJSON_AddStringToObject(pRoot, "privateTerm", u64buf); - - snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->term); - cJSON_AddStringToObject(pRoot, "term", u64buf); - - cJSON_AddStringToObject(pRoot, "matchIndex", u64buf); - snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->startTime); - cJSON_AddStringToObject(pRoot, "startTime", u64buf); + pMsg->contLen = bytes; + if (pMsg->pCont == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; } - cJSON* pJson = cJSON_CreateObject(); - cJSON_AddItemToObject(pJson, "SyncHeartbeatReply", pRoot); - return pJson; -} - -char* syncHeartbeatReply2Str(const SyncHeartbeatReply* pMsg) { - cJSON* pJson = syncHeartbeatReply2Json(pMsg); - char* serialized = cJSON_Print(pJson); - cJSON_Delete(pJson); - return serialized; -} - -void syncHeartbeatReplyPrint(const SyncHeartbeatReply* pMsg) { - char* serialized = syncHeartbeatReply2Str(pMsg); - printf("syncHeartbeatReplyPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized); - fflush(NULL); - taosMemoryFree(serialized); -} - -void syncHeartbeatReplyPrint2(char* s, const SyncHeartbeatReply* pMsg) { - char* serialized = syncHeartbeatReply2Str(pMsg); - printf("syncHeartbeatReplyPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized); - fflush(NULL); - taosMemoryFree(serialized); -} - -void syncHeartbeatReplyLog(const SyncHeartbeatReply* pMsg) { - char* serialized = syncHeartbeatReply2Str(pMsg); - sTrace("syncHeartbeatReplyLog | len:%d | %s", (int32_t)strlen(serialized), serialized); - taosMemoryFree(serialized); -} - -void syncHeartbeatReplyLog2(char* s, const SyncHeartbeatReply* pMsg) { - if (gRaftDetailLog) { - char* serialized = syncHeartbeatReply2Str(pMsg); - sTrace("syncHeartbeatReplyLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized); - taosMemoryFree(serialized); - } + SyncHeartbeatReply* pHeartbeatReply = pMsg->pCont; + pHeartbeatReply->bytes = bytes; + pHeartbeatReply->msgType = TDMT_SYNC_HEARTBEAT_REPLY; + pHeartbeatReply->vgId = vgId; + return 0; } // ---- message process SyncPreSnapshot---- diff --git a/source/libs/sync/test/sync_test_lib/inc/syncTest.h b/source/libs/sync/test/sync_test_lib/inc/syncTest.h index 4ebd78dd42..f853e7c823 100644 --- a/source/libs/sync/test/sync_test_lib/inc/syncTest.h +++ b/source/libs/sync/test/sync_test_lib/inc/syncTest.h @@ -331,6 +331,24 @@ void syncHeartbeatPrint2(char* s, const SyncHeartbeat* pMsg); void syncHeartbeatLog(const SyncHeartbeat* pMsg); void syncHeartbeatLog2(char* s, const SyncHeartbeat* pMsg); +SyncHeartbeatReply* syncHeartbeatReplyBuild(int32_t vgId); +void syncHeartbeatReplyDestroy(SyncHeartbeatReply* pMsg); +void syncHeartbeatReplySerialize(const SyncHeartbeatReply* pMsg, char* buf, uint32_t bufLen); +void syncHeartbeatReplyDeserialize(const char* buf, uint32_t len, SyncHeartbeatReply* pMsg); +char* syncHeartbeatReplySerialize2(const SyncHeartbeatReply* pMsg, uint32_t* len); +SyncHeartbeatReply* syncHeartbeatReplyDeserialize2(const char* buf, uint32_t len); +void syncHeartbeatReply2RpcMsg(const SyncHeartbeatReply* pMsg, SRpcMsg* pRpcMsg); +void syncHeartbeatReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncHeartbeatReply* pMsg); +SyncHeartbeatReply* syncHeartbeatReplyFromRpcMsg2(const SRpcMsg* pRpcMsg); +cJSON* syncHeartbeatReply2Json(const SyncHeartbeatReply* pMsg); +char* syncHeartbeatReply2Str(const SyncHeartbeatReply* pMsg); + +// for debug ---------------------- +void syncHeartbeatReplyPrint(const SyncHeartbeatReply* pMsg); +void syncHeartbeatReplyPrint2(char* s, const SyncHeartbeatReply* pMsg); +void syncHeartbeatReplyLog(const SyncHeartbeatReply* pMsg); +void syncHeartbeatReplyLog2(char* s, const SyncHeartbeatReply* pMsg); + #ifdef __cplusplus } #endif diff --git a/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c b/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c index a3bfdcb7ef..fe4f78976f 100644 --- a/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c +++ b/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c @@ -1787,4 +1787,158 @@ void syncHeartbeatLog2(char* s, const SyncHeartbeat* pMsg) { sTrace("syncHeartbeatLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized); taosMemoryFree(serialized); } -} \ No newline at end of file +} + +// ---- message process SyncHeartbeatReply---- +SyncHeartbeatReply* syncHeartbeatReplyBuild(int32_t vgId) { + uint32_t bytes = sizeof(SyncHeartbeatReply); + SyncHeartbeatReply* pMsg = taosMemoryMalloc(bytes); + memset(pMsg, 0, bytes); + pMsg->bytes = bytes; + pMsg->vgId = vgId; + pMsg->msgType = TDMT_SYNC_HEARTBEAT_REPLY; + return pMsg; +} + +void syncHeartbeatReplyDestroy(SyncHeartbeatReply* pMsg) { + if (pMsg != NULL) { + taosMemoryFree(pMsg); + } +} + +void syncHeartbeatReplySerialize(const SyncHeartbeatReply* pMsg, char* buf, uint32_t bufLen) { + ASSERT(pMsg->bytes <= bufLen); + memcpy(buf, pMsg, pMsg->bytes); +} + +void syncHeartbeatReplyDeserialize(const char* buf, uint32_t len, SyncHeartbeatReply* pMsg) { + memcpy(pMsg, buf, len); + ASSERT(len == pMsg->bytes); +} + +char* syncHeartbeatReplySerialize2(const SyncHeartbeatReply* pMsg, uint32_t* len) { + char* buf = taosMemoryMalloc(pMsg->bytes); + ASSERT(buf != NULL); + syncHeartbeatReplySerialize(pMsg, buf, pMsg->bytes); + if (len != NULL) { + *len = pMsg->bytes; + } + return buf; +} + +SyncHeartbeatReply* syncHeartbeatReplyDeserialize2(const char* buf, uint32_t len) { + uint32_t bytes = *((uint32_t*)buf); + SyncHeartbeatReply* pMsg = taosMemoryMalloc(bytes); + ASSERT(pMsg != NULL); + syncHeartbeatReplyDeserialize(buf, len, pMsg); + ASSERT(len == pMsg->bytes); + return pMsg; +} + +void syncHeartbeatReply2RpcMsg(const SyncHeartbeatReply* pMsg, SRpcMsg* pRpcMsg) { + memset(pRpcMsg, 0, sizeof(*pRpcMsg)); + pRpcMsg->msgType = pMsg->msgType; + pRpcMsg->contLen = pMsg->bytes; + pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen); + syncHeartbeatReplySerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen); +} + +void syncHeartbeatReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncHeartbeatReply* pMsg) { + syncHeartbeatReplyDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg); +} + +SyncHeartbeatReply* syncHeartbeatReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) { + SyncHeartbeatReply* pMsg = syncHeartbeatReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen); + ASSERT(pMsg != NULL); + return pMsg; +} + +cJSON* syncHeartbeatReply2Json(const SyncHeartbeatReply* pMsg) { + char u64buf[128] = {0}; + cJSON* pRoot = cJSON_CreateObject(); + + if (pMsg != NULL) { + cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes); + cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId); + cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType); + + cJSON* pSrcId = cJSON_CreateObject(); + snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->srcId.addr); + cJSON_AddStringToObject(pSrcId, "addr", u64buf); + { + uint64_t u64 = pMsg->srcId.addr; + cJSON* pTmp = pSrcId; + char host[128] = {0}; + uint16_t port; + syncUtilU642Addr(u64, host, sizeof(host), &port); + cJSON_AddStringToObject(pTmp, "addr_host", host); + cJSON_AddNumberToObject(pTmp, "addr_port", port); + } + cJSON_AddNumberToObject(pSrcId, "vgId", pMsg->srcId.vgId); + cJSON_AddItemToObject(pRoot, "srcId", pSrcId); + + cJSON* pDestId = cJSON_CreateObject(); + snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->destId.addr); + cJSON_AddStringToObject(pDestId, "addr", u64buf); + { + uint64_t u64 = pMsg->destId.addr; + cJSON* pTmp = pDestId; + char host[128] = {0}; + uint16_t port; + syncUtilU642Addr(u64, host, sizeof(host), &port); + cJSON_AddStringToObject(pTmp, "addr_host", host); + cJSON_AddNumberToObject(pTmp, "addr_port", port); + } + cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId); + cJSON_AddItemToObject(pRoot, "destId", pDestId); + + snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->privateTerm); + cJSON_AddStringToObject(pRoot, "privateTerm", u64buf); + + snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->term); + cJSON_AddStringToObject(pRoot, "term", u64buf); + + cJSON_AddStringToObject(pRoot, "matchIndex", u64buf); + snprintf(u64buf, sizeof(u64buf), "%" PRId64, pMsg->startTime); + cJSON_AddStringToObject(pRoot, "startTime", u64buf); + } + + cJSON* pJson = cJSON_CreateObject(); + cJSON_AddItemToObject(pJson, "SyncHeartbeatReply", pRoot); + return pJson; +} + +char* syncHeartbeatReply2Str(const SyncHeartbeatReply* pMsg) { + cJSON* pJson = syncHeartbeatReply2Json(pMsg); + char* serialized = cJSON_Print(pJson); + cJSON_Delete(pJson); + return serialized; +} + +void syncHeartbeatReplyPrint(const SyncHeartbeatReply* pMsg) { + char* serialized = syncHeartbeatReply2Str(pMsg); + printf("syncHeartbeatReplyPrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized); + fflush(NULL); + taosMemoryFree(serialized); +} + +void syncHeartbeatReplyPrint2(char* s, const SyncHeartbeatReply* pMsg) { + char* serialized = syncHeartbeatReply2Str(pMsg); + printf("syncHeartbeatReplyPrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized); + fflush(NULL); + taosMemoryFree(serialized); +} + +void syncHeartbeatReplyLog(const SyncHeartbeatReply* pMsg) { + char* serialized = syncHeartbeatReply2Str(pMsg); + sTrace("syncHeartbeatReplyLog | len:%d | %s", (int32_t)strlen(serialized), serialized); + taosMemoryFree(serialized); +} + +void syncHeartbeatReplyLog2(char* s, const SyncHeartbeatReply* pMsg) { + if (gRaftDetailLog) { + char* serialized = syncHeartbeatReply2Str(pMsg); + sTrace("syncHeartbeatReplyLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized); + taosMemoryFree(serialized); + } +}