sync io

source/libs/sync/src/syncIO.c

@@ -112,7 +112,7 @@ static int32_t syncIOStartInternal(SSyncIO *io) {
   {
     SRpcInit rpcInit;
     memset(&rpcInit, 0, sizeof(rpcInit));
-    rpcInit.localPort = 7010;
+    rpcInit.localPort = io->myAddr.eps[0].port;
     rpcInit.label = "SYNC-IO-SERVER";
     rpcInit.numOfThreads = 1;
     rpcInit.cfp = syncIOProcessRequest;

source/libs/sync/test/CMakeLists.txt

@@ -4,6 +4,9 @@ add_executable(syncPingTest "")
 add_executable(syncEncodeTest "")
 add_executable(syncIOTickQTest "")
 add_executable(syncIOTickPingTest "")
+add_executable(syncIOSendMsgTest "")
+add_executable(syncIOSendMsgClientTest "")
+add_executable(syncIOSendMsgServerTest "")
 
 
 target_sources(syncTest
@@ -30,6 +33,18 @@ target_sources(syncIOTickPingTest
     PRIVATE
     "syncIOTickPingTest.cpp"
 )
+target_sources(syncIOSendMsgTest
+    PRIVATE
+    "syncIOSendMsgTest.cpp"
+)
+target_sources(syncIOSendMsgClientTest
+    PRIVATE
+    "syncIOSendMsgClientTest.cpp"
+)
+target_sources(syncIOSendMsgServerTest
+    PRIVATE
+    "syncIOSendMsgServerTest.cpp"
+)
 
 
 target_include_directories(syncTest
@@ -62,6 +77,21 @@ target_include_directories(syncIOTickPingTest
     "${CMAKE_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
+target_include_directories(syncIOSendMsgTest
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
+target_include_directories(syncIOSendMsgClientTest
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
+target_include_directories(syncIOSendMsgServerTest
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
 
 
 target_link_libraries(syncTest
@@ -88,6 +118,18 @@ target_link_libraries(syncIOTickPingTest
     sync
     gtest_main
 )
+target_link_libraries(syncIOSendMsgTest
+    sync
+    gtest_main
+)
+target_link_libraries(syncIOSendMsgClientTest
+    sync
+    gtest_main
+)
+target_link_libraries(syncIOSendMsgServerTest
+    sync
+    gtest_main
+)
 
 
 enable_testing()

source/libs/sync/test/syncIOSendMsgClientTest.cpp

+#include <stdio.h>
+#include "gtest/gtest.h"
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncRaftStore.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");
+}
+
+int main() {
+  // taosInitLog((char *)"syncTest.log", 100000, 10);
+  tsAsyncLog = 0;
+  sDebugFlag = 143 + 64;
+
+  logTest();
+
+  int32_t ret;
+
+  ret = syncIOStart((char *)"127.0.0.1", 7010);
+  assert(ret == 0);
+
+  for (int i = 0; i < 10; ++i) {
+    SEpSet epSet;
+    epSet.inUse = 0;
+    epSet.numOfEps = 1;
+    addEpIntoEpSet(&epSet, "127.0.0.1", 7030);
+
+    SRpcMsg rpcMsg;
+    rpcMsg.contLen = 64;
+    rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
+    snprintf((char*)rpcMsg.pCont, rpcMsg.contLen, "%s", "syncIOSendMsgTest");
+    rpcMsg.handle = NULL;
+    rpcMsg.msgType = 77;
+
+    syncIOSendMsg(gSyncIO->clientRpc, &epSet, &rpcMsg);
+    sleep(1);
+  }
+
+  while (1) {
+    sleep(1);
+  }
+
+  return 0;
+}

source/libs/sync/test/syncIOSendMsgServerTest.cpp

+#include <stdio.h>
+#include "gtest/gtest.h"
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncRaftStore.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");
+}
+
+int main() {
+  // taosInitLog((char *)"syncTest.log", 100000, 10);
+  tsAsyncLog = 0;
+  sDebugFlag = 143 + 64;
+
+  logTest();
+
+  int32_t ret;
+
+  ret = syncIOStart((char *)"127.0.0.1", 7030);
+  assert(ret == 0);
+
+  while (1) {
+    sleep(1);
+  }
+
+  return 0;
+}

source/libs/sync/test/syncIOSendMsgTest.cpp

+#include <stdio.h>
+#include "gtest/gtest.h"
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncRaftStore.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");
+}
+
+int main() {
+  // taosInitLog((char *)"syncTest.log", 100000, 10);
+  tsAsyncLog = 0;
+  sDebugFlag = 143 + 64;
+
+  logTest();
+
+  int32_t ret;
+
+  ret = syncIOStart((char *)"127.0.0.1", 7010);
+  assert(ret == 0);
+
+  for (int i = 0; i < 10; ++i) {
+    SEpSet epSet;
+    epSet.inUse = 0;
+    epSet.numOfEps = 1;
+    addEpIntoEpSet(&epSet, "127.0.0.1", 7010);
+
+    SRpcMsg rpcMsg;
+    rpcMsg.contLen = 64;
+    rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
+    snprintf((char*)rpcMsg.pCont, rpcMsg.contLen, "%s", "syncIOSendMsgTest");
+    rpcMsg.handle = NULL;
+    rpcMsg.msgType = 77;
+
+    syncIOSendMsg(gSyncIO->clientRpc, &epSet, &rpcMsg);
+    sleep(1);
+  }
+
+  while (1) {
+    sleep(1);
+  }
+
+  return 0;
+}

source/libs/sync/test/syncPingTest.cpp

@@ -68,10 +68,11 @@ int main() {
   ret = syncNodeStartPingTimer(pSyncNode);
   assert(ret == 0);
 
-  taosMsleep(5000);
-
+  /*
+  taosMsleep(10000);
   ret = syncNodeStopPingTimer(pSyncNode);
   assert(ret == 0);
+  */
 
   while (1) {
     taosMsleep(1000);