add test sender, receiver

source/libs/sync/inc/syncSnapshot.h

@@ -37,8 +37,8 @@ typedef struct SSyncSnapshotSender {
   bool       start;
   int32_t    seq;
   int32_t    ack;
-  void      *pReader;
-  void      *pCurrentBlock;
+  void *     pReader;
+  void *     pCurrentBlock;
   int32_t    blockLen;
   SSnapshot  snapshot;
   int64_t    sendingMS;
@@ -52,15 +52,15 @@ void                 snapshotSenderDestroy(SSyncSnapshotSender *pSender);
 void                 snapshotSenderStart(SSyncSnapshotSender *pSender);
 void                 snapshotSenderStop(SSyncSnapshotSender *pSender);
 int32_t              snapshotSend(SSyncSnapshotSender *pSender);
-cJSON               *snapshotSender2Json(SSyncSnapshotSender *pSender);
-char                *snapshotSender2Str(SSyncSnapshotSender *pSender);
+cJSON *              snapshotSender2Json(SSyncSnapshotSender *pSender);
+char *               snapshotSender2Str(SSyncSnapshotSender *pSender);
 
 typedef struct SSyncSnapshotReceiver {
   bool start;
 
   int32_t  ack;
-  void    *pWriter;
-  void    *pCurrentBlock;
+  void *   pWriter;
+  void *   pCurrentBlock;
   int32_t  blockLen;
   SyncTerm term;
 
@@ -72,8 +72,8 @@ SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, int32_t repl
 void                   snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver);
 void                   snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver);
 void                   snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver);
-cJSON                 *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver);
-char                  *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver);
+cJSON *                snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver);
+char *                 snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver);
 
 int32_t syncNodeOnSnapshotSendCb(SSyncNode *ths, SyncSnapshotSend *pMsg);
 int32_t syncNodeOnSnapshotRspCb(SSyncNode *ths, SyncSnapshotRsp *pMsg);

source/libs/sync/src/syncSnapshot.c

@@ -175,9 +175,9 @@ cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) {
 
     cJSON *pSnapshot = cJSON_CreateObject();
     snprintf(u64buf, sizeof(u64buf), "%lu", pSender->snapshot.lastApplyIndex);
-    cJSON_AddStringToObject(pRoot, "lastApplyIndex", u64buf);
+    cJSON_AddStringToObject(pSnapshot, "lastApplyIndex", u64buf);
     snprintf(u64buf, sizeof(u64buf), "%lu", pSender->snapshot.lastApplyTerm);
-    cJSON_AddStringToObject(pRoot, "lastApplyTerm", u64buf);
+    cJSON_AddStringToObject(pSnapshot, "lastApplyTerm", u64buf);
     cJSON_AddItemToObject(pRoot, "snapshot", pSnapshot);
 
     snprintf(u64buf, sizeof(u64buf), "%lu", pSender->sendingMS);

source/libs/sync/test/CMakeLists.txt

@@ -40,6 +40,8 @@ add_executable(syncApplyMsgTest "")
 add_executable(syncConfigChangeTest "")
 add_executable(syncSnapshotSendTest "")
 add_executable(syncSnapshotRspTest "")
+add_executable(syncSnapshotSenderTest "")
+add_executable(syncSnapshotReceiverTest "")
 
 
 target_sources(syncTest
@@ -210,6 +212,14 @@ target_sources(syncSnapshotRspTest
     PRIVATE
     "syncSnapshotRspTest.cpp"
 )
+target_sources(syncSnapshotSenderTest
+    PRIVATE
+    "syncSnapshotSenderTest.cpp"
+)
+target_sources(syncSnapshotReceiverTest
+    PRIVATE
+    "syncSnapshotReceiverTest.cpp"
+)
 
 
 target_include_directories(syncTest
@@ -422,6 +432,16 @@ target_include_directories(syncSnapshotRspTest
     "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
+target_include_directories(syncSnapshotSenderTest
+    PUBLIC
+    "${TD_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
+target_include_directories(syncSnapshotReceiverTest
+    PUBLIC
+    "${TD_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
 
 
 target_link_libraries(syncTest
@@ -592,6 +612,14 @@ target_link_libraries(syncSnapshotRspTest
     sync
     gtest_main
 )
+target_link_libraries(syncSnapshotSenderTest
+    sync
+    gtest_main
+)
+target_link_libraries(syncSnapshotReceiverTest
+    sync
+    gtest_main
+)
 
 
 enable_testing()

source/libs/sync/test/syncSnapshotReceiverTest.cpp

+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncMessage.h"
+#include "syncRaftStore.h"
+#include "syncSnapshot.h"
+#include "syncUtil.h"
+
+void logTest() {
+  sTrace("--- sync log test: trace");
+  sDebug("--- sync log test: debug");
+  sInfo("--- sync log test: info");
+  sWarn("--- sync log test: warn");
+  sError("--- sync log test: error");
+  sFatal("--- sync log test: fatal");
+}
+
+void CommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {}
+void PreCommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {}
+void RollBackCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {}
+
+void RestoreFinishCb(struct SSyncFSM* pFsm) {}
+void ReConfigCb(struct SSyncFSM* pFsm, SSyncCfg newCfg, SReConfigCbMeta cbMeta) {}
+
+int32_t GetSnapshot(struct SSyncFSM* pFsm, SSnapshot* pSnapshot) { return 0; }
+
+int32_t SnapshotStartRead(struct SSyncFSM* pFsm, void** ppReader) { return 0; }
+int32_t SnapshotStopRead(struct SSyncFSM* pFsm, void* pReader) { return 0; }
+int32_t SnapshotDoRead(struct SSyncFSM* pFsm, void* pReader, void** ppBuf, int32_t* len) { return 0; }
+
+int32_t SnapshotStartWrite(struct SSyncFSM* pFsm, void** ppWriter) { return 0; }
+int32_t SnapshotStopWrite(struct SSyncFSM* pFsm, void* pWriter, bool isApply) { return 0; }
+int32_t SnapshotDoWrite(struct SSyncFSM* pFsm, void* pWriter, void* pBuf, int32_t len) { return 0; }
+
+SSyncSnapshotReceiver* createReceiver() {
+  SSyncNode* pSyncNode = (SSyncNode*)taosMemoryMalloc(sizeof(*pSyncNode));
+  pSyncNode->pRaftStore = (SRaftStore*)taosMemoryMalloc(sizeof(*(pSyncNode->pRaftStore)));
+  pSyncNode->pFsm = (SSyncFSM*)taosMemoryMalloc(sizeof(*(pSyncNode->pFsm)));
+  pSyncNode->pFsm->FpSnapshotStartWrite = SnapshotStartWrite;
+  pSyncNode->pFsm->FpSnapshotStopWrite = SnapshotStopWrite;
+  pSyncNode->pFsm->FpSnapshotDoWrite = SnapshotDoWrite;
+
+  SSyncSnapshotReceiver* pReceiver = snapshotReceiverCreate(pSyncNode, 2);
+  pReceiver->start = true;
+  pReceiver->ack = 20;
+  pReceiver->pWriter = (void*)0x11;
+  pReceiver->blockLen = 20;
+  pReceiver->pCurrentBlock = taosMemoryMalloc(pReceiver->blockLen);
+  snprintf((char*)(pReceiver->pCurrentBlock), pReceiver->blockLen, "%s", "hello");
+  pReceiver->term = 66;
+
+  return pReceiver;
+}
+
+int main() {
+  tsAsyncLog = 0;
+  sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
+  logTest();
+
+  SSyncSnapshotReceiver* pReceiver = createReceiver();
+  sTrace("%s", snapshotReceiver2Str(pReceiver));
+
+  return 0;
+}

source/libs/sync/test/syncSnapshotSenderTest.cpp

+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncMessage.h"
+#include "syncRaftStore.h"
+#include "syncSnapshot.h"
+#include "syncUtil.h"
+
+void logTest() {
+  sTrace("--- sync log test: trace");
+  sDebug("--- sync log test: debug");
+  sInfo("--- sync log test: info");
+  sWarn("--- sync log test: warn");
+  sError("--- sync log test: error");
+  sFatal("--- sync log test: fatal");
+}
+
+void CommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {}
+void PreCommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {}
+void RollBackCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {}
+
+void RestoreFinishCb(struct SSyncFSM* pFsm) {}
+void ReConfigCb(struct SSyncFSM* pFsm, SSyncCfg newCfg, SReConfigCbMeta cbMeta) {}
+
+int32_t GetSnapshot(struct SSyncFSM* pFsm, SSnapshot* pSnapshot) { return 0; }
+
+int32_t SnapshotStartRead(struct SSyncFSM* pFsm, void** ppReader) { return 0; }
+int32_t SnapshotStopRead(struct SSyncFSM* pFsm, void* pReader) { return 0; }
+int32_t SnapshotDoRead(struct SSyncFSM* pFsm, void* pReader, void** ppBuf, int32_t* len) { return 0; }
+
+int32_t SnapshotStartWrite(struct SSyncFSM* pFsm, void** ppWriter) { return 0; }
+int32_t SnapshotStopWrite(struct SSyncFSM* pFsm, void* pWriter, bool isApply) { return 0; }
+int32_t SnapshotDoWrite(struct SSyncFSM* pFsm, void* pWriter, void* pBuf, int32_t len) { return 0; }
+
+SSyncSnapshotSender* createSender() {
+  SSyncNode* pSyncNode = (SSyncNode*)taosMemoryMalloc(sizeof(*pSyncNode));
+  pSyncNode->pRaftStore = (SRaftStore*)taosMemoryMalloc(sizeof(*(pSyncNode->pRaftStore)));
+  pSyncNode->pFsm = (SSyncFSM*)taosMemoryMalloc(sizeof(*(pSyncNode->pFsm)));
+  pSyncNode->pFsm->FpSnapshotStartRead = SnapshotStartRead;
+  pSyncNode->pFsm->FpSnapshotStopRead = SnapshotStopRead;
+  pSyncNode->pFsm->FpSnapshotDoRead = SnapshotDoRead;
+
+  SSyncSnapshotSender* pSender = snapshotSenderCreate(pSyncNode, 2);
+  pSender->start = true;
+  pSender->seq = 10;
+  pSender->ack = 20;
+  pSender->pReader = (void*)0x11;
+  pSender->blockLen = 20;
+  pSender->pCurrentBlock = taosMemoryMalloc(pSender->blockLen);
+  snprintf((char*)(pSender->pCurrentBlock), pSender->blockLen, "%s", "hello");
+
+  pSender->snapshot.lastApplyIndex = 99;
+  pSender->snapshot.lastApplyTerm = 88;
+  pSender->sendingMS = 77;
+  pSender->term = 66;
+
+  return pSender;
+}
+
+int main() {
+  tsAsyncLog = 0;
+  sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
+  logTest();
+
+  SSyncSnapshotSender* pSender = createSender();
+  sTrace("%s", snapshotSender2Str(pSender));
+
+  return 0;
+}