syncMessage.c 54.5 KB
Newer Older
M
Minghao Li 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "syncMessage.h"
M
Minghao Li 已提交
17
#include "syncUtil.h"
M
Minghao Li 已提交
18
#include "tcoding.h"
M
Minghao Li 已提交
19

M
Minghao Li 已提交
20 21 22 23
// ---------------------------------------------
cJSON* syncRpcMsg2Json(SRpcMsg* pRpcMsg) {
  cJSON* pRoot;

M
Minghao Li 已提交
24
  // in compiler optimization, switch case = if else constants
M
Minghao Li 已提交
25
  if (pRpcMsg->msgType == TDMT_VND_SYNC_TIMEOUT) {
M
Minghao Li 已提交
26
    SyncTimeout* pSyncMsg = syncTimeoutDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
M
Minghao Li 已提交
27
    pRoot = syncTimeout2Json(pSyncMsg);
M
Minghao Li 已提交
28
    syncTimeoutDestroy(pSyncMsg);
M
Minghao Li 已提交
29

M
Minghao Li 已提交
30
  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING) {
M
Minghao Li 已提交
31
    SyncPing* pSyncMsg = syncPingDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
M
Minghao Li 已提交
32
    pRoot = syncPing2Json(pSyncMsg);
M
Minghao Li 已提交
33
    syncPingDestroy(pSyncMsg);
M
Minghao Li 已提交
34

M
Minghao Li 已提交
35
  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING_REPLY) {
M
Minghao Li 已提交
36
    SyncPingReply* pSyncMsg = syncPingReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
M
Minghao Li 已提交
37
    pRoot = syncPingReply2Json(pSyncMsg);
M
Minghao Li 已提交
38
    syncPingReplyDestroy(pSyncMsg);
M
Minghao Li 已提交
39

M
Minghao Li 已提交
40
  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_CLIENT_REQUEST) {
M
Minghao Li 已提交
41
    SyncClientRequest* pSyncMsg = syncClientRequestDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
M
Minghao Li 已提交
42
    pRoot = syncClientRequest2Json(pSyncMsg);
M
Minghao Li 已提交
43
    syncClientRequestDestroy(pSyncMsg);
M
Minghao Li 已提交
44

M
Minghao Li 已提交
45
  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_CLIENT_REQUEST_REPLY) {
M
Minghao Li 已提交
46 47
    pRoot = syncRpcUnknownMsg2Json();

M
Minghao Li 已提交
48
  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE) {
M
Minghao Li 已提交
49
    SyncRequestVote* pSyncMsg = syncRequestVoteDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
M
Minghao Li 已提交
50
    pRoot = syncRequestVote2Json(pSyncMsg);
M
Minghao Li 已提交
51
    syncRequestVoteDestroy(pSyncMsg);
M
Minghao Li 已提交
52

M
Minghao Li 已提交
53
  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE_REPLY) {
M
Minghao Li 已提交
54
    SyncRequestVoteReply* pSyncMsg = syncRequestVoteReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
M
Minghao Li 已提交
55
    pRoot = syncRequestVoteReply2Json(pSyncMsg);
M
Minghao Li 已提交
56
    syncRequestVoteReplyDestroy(pSyncMsg);
M
Minghao Li 已提交
57

M
Minghao Li 已提交
58
  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES) {
M
Minghao Li 已提交
59
    SyncAppendEntries* pSyncMsg = syncAppendEntriesDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
M
Minghao Li 已提交
60
    pRoot = syncAppendEntries2Json(pSyncMsg);
M
Minghao Li 已提交
61
    syncAppendEntriesDestroy(pSyncMsg);
M
Minghao Li 已提交
62

M
Minghao Li 已提交
63
  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES_REPLY) {
M
Minghao Li 已提交
64
    SyncAppendEntriesReply* pSyncMsg = syncAppendEntriesReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
M
Minghao Li 已提交
65
    pRoot = syncAppendEntriesReply2Json(pSyncMsg);
M
Minghao Li 已提交
66
    syncAppendEntriesReplyDestroy(pSyncMsg);
M
Minghao Li 已提交
67

M
Minghao Li 已提交
68
  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_COMMON_RESPONSE) {
M
Minghao Li 已提交
69 70 71 72
    pRoot = cJSON_CreateObject();
    char* s;
    s = syncUtilprintBin((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
    cJSON_AddStringToObject(pRoot, "pCont", s);
wafwerar's avatar
wafwerar 已提交
73
    taosMemoryFree(s);
M
Minghao Li 已提交
74 75
    s = syncUtilprintBin2((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
    cJSON_AddStringToObject(pRoot, "pCont2", s);
wafwerar's avatar
wafwerar 已提交
76
    taosMemoryFree(s);
M
Minghao Li 已提交
77

M
Minghao Li 已提交
78
  } else {
M
Minghao Li 已提交
79
    pRoot = cJSON_CreateObject();
M
Minghao Li 已提交
80 81 82
    char* s;
    s = syncUtilprintBin((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
    cJSON_AddStringToObject(pRoot, "pCont", s);
wafwerar's avatar
wafwerar 已提交
83
    taosMemoryFree(s);
M
Minghao Li 已提交
84 85
    s = syncUtilprintBin2((char*)(pRpcMsg->pCont), pRpcMsg->contLen);
    cJSON_AddStringToObject(pRoot, "pCont2", s);
wafwerar's avatar
wafwerar 已提交
86
    taosMemoryFree(s);
M
Minghao Li 已提交
87 88
  }

M
Minghao Li 已提交
89 90 91
  cJSON_AddNumberToObject(pRoot, "msgType", pRpcMsg->msgType);
  cJSON_AddNumberToObject(pRoot, "contLen", pRpcMsg->contLen);
  cJSON_AddNumberToObject(pRoot, "code", pRpcMsg->code);
M
Minghao Li 已提交
92
  // cJSON_AddNumberToObject(pRoot, "persist", pRpcMsg->persist);
M
Minghao Li 已提交
93

M
Minghao Li 已提交
94 95 96 97 98 99 100
  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "RpcMsg", pRoot);
  return pJson;
}

cJSON* syncRpcUnknownMsg2Json() {
  cJSON* pRoot = cJSON_CreateObject();
M
Minghao Li 已提交
101
  cJSON_AddNumberToObject(pRoot, "msgType", TDMT_VND_SYNC_UNKNOWN);
M
Minghao Li 已提交
102
  cJSON_AddStringToObject(pRoot, "data", "unknown message");
M
Minghao Li 已提交
103 104

  cJSON* pJson = cJSON_CreateObject();
M
Minghao Li 已提交
105
  cJSON_AddItemToObject(pJson, "SyncUnknown", pRoot);
M
Minghao Li 已提交
106 107 108
  return pJson;
}

M
Minghao Li 已提交
109 110 111 112 113 114 115
char* syncRpcMsg2Str(SRpcMsg* pRpcMsg) {
  cJSON* pJson = syncRpcMsg2Json(pRpcMsg);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}

M
Minghao Li 已提交
116 117 118 119 120
// for debug ----------------------
void syncRpcMsgPrint(SRpcMsg* pMsg) {
  char* serialized = syncRpcMsg2Str(pMsg);
  printf("syncRpcMsgPrint | len:%lu | %s \n", strlen(serialized), serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
121
  taosMemoryFree(serialized);
M
Minghao Li 已提交
122 123 124 125 126 127
}

void syncRpcMsgPrint2(char* s, SRpcMsg* pMsg) {
  char* serialized = syncRpcMsg2Str(pMsg);
  printf("syncRpcMsgPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
128
  taosMemoryFree(serialized);
M
Minghao Li 已提交
129 130 131 132 133
}

void syncRpcMsgLog(SRpcMsg* pMsg) {
  char* serialized = syncRpcMsg2Str(pMsg);
  sTrace("syncRpcMsgLog | len:%lu | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
134
  taosMemoryFree(serialized);
M
Minghao Li 已提交
135 136 137 138 139
}

void syncRpcMsgLog2(char* s, SRpcMsg* pMsg) {
  char* serialized = syncRpcMsg2Str(pMsg);
  sTrace("syncRpcMsgLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
wafwerar's avatar
wafwerar 已提交
140
  taosMemoryFree(serialized);
M
Minghao Li 已提交
141 142
}

M
Minghao Li 已提交
143 144 145
// ---- message process SyncTimeout----
SyncTimeout* syncTimeoutBuild() {
  uint32_t     bytes = sizeof(SyncTimeout);
wafwerar's avatar
wafwerar 已提交
146
  SyncTimeout* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
147 148
  memset(pMsg, 0, bytes);
  pMsg->bytes = bytes;
M
Minghao Li 已提交
149
  pMsg->msgType = TDMT_VND_SYNC_TIMEOUT;
M
Minghao Li 已提交
150 151 152
  return pMsg;
}

M
Minghao Li 已提交
153 154
SyncTimeout* syncTimeoutBuild2(ESyncTimeoutType timeoutType, uint64_t logicClock, int32_t timerMS, int32_t vgId,
                               void* data) {
M
Minghao Li 已提交
155
  SyncTimeout* pMsg = syncTimeoutBuild();
M
Minghao Li 已提交
156
  pMsg->vgId = vgId;
M
Minghao Li 已提交
157 158 159 160 161 162 163
  pMsg->timeoutType = timeoutType;
  pMsg->logicClock = logicClock;
  pMsg->timerMS = timerMS;
  pMsg->data = data;
  return pMsg;
}

M
Minghao Li 已提交
164 165
void syncTimeoutDestroy(SyncTimeout* pMsg) {
  if (pMsg != NULL) {
wafwerar's avatar
wafwerar 已提交
166
    taosMemoryFree(pMsg);
M
Minghao Li 已提交
167 168 169 170 171 172 173 174 175 176 177 178 179
  }
}

void syncTimeoutSerialize(const SyncTimeout* pMsg, char* buf, uint32_t bufLen) {
  assert(pMsg->bytes <= bufLen);
  memcpy(buf, pMsg, pMsg->bytes);
}

void syncTimeoutDeserialize(const char* buf, uint32_t len, SyncTimeout* pMsg) {
  memcpy(pMsg, buf, len);
  assert(len == pMsg->bytes);
}

M
Minghao Li 已提交
180
char* syncTimeoutSerialize2(const SyncTimeout* pMsg, uint32_t* len) {
wafwerar's avatar
wafwerar 已提交
181
  char* buf = taosMemoryMalloc(pMsg->bytes);
M
Minghao Li 已提交
182 183 184 185 186 187 188 189 190 191
  assert(buf != NULL);
  syncTimeoutSerialize(pMsg, buf, pMsg->bytes);
  if (len != NULL) {
    *len = pMsg->bytes;
  }
  return buf;
}

SyncTimeout* syncTimeoutDeserialize2(const char* buf, uint32_t len) {
  uint32_t     bytes = *((uint32_t*)buf);
wafwerar's avatar
wafwerar 已提交
192
  SyncTimeout* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
193 194 195 196 197 198
  assert(pMsg != NULL);
  syncTimeoutDeserialize(buf, len, pMsg);
  assert(len == pMsg->bytes);
  return pMsg;
}

M
Minghao Li 已提交
199 200 201 202 203 204 205 206 207 208 209 210
void syncTimeout2RpcMsg(const SyncTimeout* pMsg, SRpcMsg* pRpcMsg) {
  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
  pRpcMsg->msgType = pMsg->msgType;
  pRpcMsg->contLen = pMsg->bytes;
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
  syncTimeoutSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
}

void syncTimeoutFromRpcMsg(const SRpcMsg* pRpcMsg, SyncTimeout* pMsg) {
  syncTimeoutDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}

M
Minghao Li 已提交
211 212 213 214 215
SyncTimeout* syncTimeoutFromRpcMsg2(const SRpcMsg* pRpcMsg) {
  SyncTimeout* pMsg = syncTimeoutDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
  return pMsg;
}

M
Minghao Li 已提交
216
cJSON* syncTimeout2Json(const SyncTimeout* pMsg) {
M
Minghao Li 已提交
217
  char   u64buf[128];
M
Minghao Li 已提交
218
  cJSON* pRoot = cJSON_CreateObject();
M
Minghao Li 已提交
219 220 221

  if (pMsg != NULL) {
    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
M
Minghao Li 已提交
222
    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
M
Minghao Li 已提交
223 224 225 226 227 228 229 230
    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
    cJSON_AddNumberToObject(pRoot, "timeoutType", pMsg->timeoutType);
    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->logicClock);
    cJSON_AddStringToObject(pRoot, "logicClock", u64buf);
    cJSON_AddNumberToObject(pRoot, "timerMS", pMsg->timerMS);
    snprintf(u64buf, sizeof(u64buf), "%p", pMsg->data);
    cJSON_AddStringToObject(pRoot, "data", u64buf);
  }
M
Minghao Li 已提交
231 232 233 234 235 236

  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SyncTimeout", pRoot);
  return pJson;
}

M
Minghao Li 已提交
237 238 239 240 241 242 243 244 245 246
char* syncTimeout2Str(const SyncTimeout* pMsg) {
  cJSON* pJson = syncTimeout2Json(pMsg);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}

// for debug ----------------------
void syncTimeoutPrint(const SyncTimeout* pMsg) {
  char* serialized = syncTimeout2Str(pMsg);
247
  printf("syncTimeoutPrint | len:%zu | %s \n", strlen(serialized), serialized);
M
Minghao Li 已提交
248
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
249
  taosMemoryFree(serialized);
M
Minghao Li 已提交
250 251 252 253 254 255
}

void syncTimeoutPrint2(char* s, const SyncTimeout* pMsg) {
  char* serialized = syncTimeout2Str(pMsg);
  printf("syncTimeoutPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
256
  taosMemoryFree(serialized);
M
Minghao Li 已提交
257 258 259 260 261
}

void syncTimeoutLog(const SyncTimeout* pMsg) {
  char* serialized = syncTimeout2Str(pMsg);
  sTrace("syncTimeoutLog | len:%lu | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
262
  taosMemoryFree(serialized);
M
Minghao Li 已提交
263 264 265 266 267
}

void syncTimeoutLog2(char* s, const SyncTimeout* pMsg) {
  char* serialized = syncTimeout2Str(pMsg);
  sTrace("syncTimeoutLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
wafwerar's avatar
wafwerar 已提交
268
  taosMemoryFree(serialized);
M
Minghao Li 已提交
269 270
}

M
Minghao Li 已提交
271
// ---- message process SyncPing----
M
Minghao Li 已提交
272
SyncPing* syncPingBuild(uint32_t dataLen) {
M
Minghao Li 已提交
273
  uint32_t  bytes = sizeof(SyncPing) + dataLen;
wafwerar's avatar
wafwerar 已提交
274
  SyncPing* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
275 276
  memset(pMsg, 0, bytes);
  pMsg->bytes = bytes;
M
Minghao Li 已提交
277
  pMsg->msgType = TDMT_VND_SYNC_PING;
M
Minghao Li 已提交
278
  pMsg->dataLen = dataLen;
M
Minghao Li 已提交
279
  return pMsg;
M
Minghao Li 已提交
280 281
}

M
Minghao Li 已提交
282
SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, int32_t vgId, const char* str) {
M
Minghao Li 已提交
283 284
  uint32_t  dataLen = strlen(str) + 1;
  SyncPing* pMsg = syncPingBuild(dataLen);
M
Minghao Li 已提交
285
  pMsg->vgId = vgId;
M
Minghao Li 已提交
286 287 288 289 290 291
  pMsg->srcId = *srcId;
  pMsg->destId = *destId;
  snprintf(pMsg->data, pMsg->dataLen, "%s", str);
  return pMsg;
}

M
Minghao Li 已提交
292 293
SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId, int32_t vgId) {
  SyncPing* pMsg = syncPingBuild2(srcId, destId, vgId, "ping");
M
Minghao Li 已提交
294 295 296
  return pMsg;
}

M
Minghao Li 已提交
297 298
void syncPingDestroy(SyncPing* pMsg) {
  if (pMsg != NULL) {
wafwerar's avatar
wafwerar 已提交
299
    taosMemoryFree(pMsg);
M
Minghao Li 已提交
300 301 302
  }
}

M
Minghao Li 已提交
303 304 305
void syncPingSerialize(const SyncPing* pMsg, char* buf, uint32_t bufLen) {
  assert(pMsg->bytes <= bufLen);
  memcpy(buf, pMsg, pMsg->bytes);
M
Minghao Li 已提交
306 307
}

M
Minghao Li 已提交
308 309 310
void syncPingDeserialize(const char* buf, uint32_t len, SyncPing* pMsg) {
  memcpy(pMsg, buf, len);
  assert(len == pMsg->bytes);
M
Minghao Li 已提交
311
  assert(pMsg->bytes == sizeof(SyncPing) + pMsg->dataLen);
M
Minghao Li 已提交
312 313
}

M
Minghao Li 已提交
314
char* syncPingSerialize2(const SyncPing* pMsg, uint32_t* len) {
wafwerar's avatar
wafwerar 已提交
315
  char* buf = taosMemoryMalloc(pMsg->bytes);
M
Minghao Li 已提交
316 317 318 319 320 321 322 323 324 325
  assert(buf != NULL);
  syncPingSerialize(pMsg, buf, pMsg->bytes);
  if (len != NULL) {
    *len = pMsg->bytes;
  }
  return buf;
}

SyncPing* syncPingDeserialize2(const char* buf, uint32_t len) {
  uint32_t  bytes = *((uint32_t*)buf);
wafwerar's avatar
wafwerar 已提交
326
  SyncPing* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
327 328 329 330 331 332
  assert(pMsg != NULL);
  syncPingDeserialize(buf, len, pMsg);
  assert(len == pMsg->bytes);
  return pMsg;
}

M
Minghao Li 已提交
333
int32_t syncPingSerialize3(const SyncPing* pMsg, char* buf, int32_t bufLen) {
H
Hongze Cheng 已提交
334 335
  SEncoder encoder = {0};
  tEncoderInit(&encoder, buf, bufLen);
M
Minghao Li 已提交
336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369
  if (tStartEncode(&encoder) < 0) {
    return -1;
  }

  if (tEncodeU32(&encoder, pMsg->bytes) < 0) {
    return -1;
  }
  if (tEncodeI32(&encoder, pMsg->vgId) < 0) {
    return -1;
  }
  if (tEncodeU32(&encoder, pMsg->msgType) < 0) {
    return -1;
  }
  if (tEncodeU64(&encoder, pMsg->srcId.addr) < 0) {
    return -1;
  }
  if (tEncodeI32(&encoder, pMsg->srcId.vgId) < 0) {
    return -1;
  }
  if (tEncodeU64(&encoder, pMsg->destId.addr) < 0) {
    return -1;
  }
  if (tEncodeI32(&encoder, pMsg->destId.vgId) < 0) {
    return -1;
  }
  if (tEncodeU32(&encoder, pMsg->dataLen) < 0) {
    return -1;
  }
  if (tEncodeBinary(&encoder, pMsg->data, pMsg->dataLen)) {
    return -1;
  }

  tEndEncode(&encoder);
  int32_t tlen = encoder.pos;
H
Hongze Cheng 已提交
370
  tEncoderClear(&encoder);
M
Minghao Li 已提交
371 372 373 374
  return tlen;
}

SyncPing* syncPingDeserialize3(void* buf, int32_t bufLen) {
H
Hongze Cheng 已提交
375 376
  SDecoder decoder = {0};
  tDecoderInit(&decoder, buf, bufLen);
M
Minghao Li 已提交
377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411
  if (tStartDecode(&decoder) < 0) {
    return NULL;
  }

  SyncPing* pMsg = NULL;
  uint32_t  bytes;
  if (tDecodeU32(&decoder, &bytes) < 0) {
    return NULL;
  }

  pMsg = taosMemoryMalloc(bytes);
  assert(pMsg != NULL);
  pMsg->bytes = bytes;

  if (tDecodeI32(&decoder, &pMsg->vgId) < 0) {
    return NULL;
  }
  if (tDecodeU32(&decoder, &pMsg->msgType) < 0) {
    return NULL;
  }
  if (tDecodeU64(&decoder, &pMsg->srcId.addr) < 0) {
    return NULL;
  }
  if (tDecodeI32(&decoder, &pMsg->srcId.vgId) < 0) {
    return NULL;
  }
  if (tDecodeU64(&decoder, &pMsg->destId.addr) < 0) {
    return NULL;
  }
  if (tDecodeI32(&decoder, &pMsg->destId.vgId) < 0) {
    return NULL;
  }
  if (tDecodeU32(&decoder, &pMsg->dataLen) < 0) {
    return NULL;
  }
H
Hongze Cheng 已提交
412
  uint32_t len;
M
Minghao Li 已提交
413
  char*    data = NULL;
H
Hongze Cheng 已提交
414
  if (tDecodeBinary(&decoder, (uint8_t**)(&data), &len) < 0) {
M
Minghao Li 已提交
415 416 417 418 419 420
    return NULL;
  }
  assert(len = pMsg->dataLen);
  memcpy(pMsg->data, data, len);

  tEndDecode(&decoder);
H
Hongze Cheng 已提交
421
  tDecoderClear(&decoder);
M
Minghao Li 已提交
422 423 424
  return pMsg;
}

M
Minghao Li 已提交
425
void syncPing2RpcMsg(const SyncPing* pMsg, SRpcMsg* pRpcMsg) {
M
Minghao Li 已提交
426
  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
M
Minghao Li 已提交
427 428
  pRpcMsg->msgType = pMsg->msgType;
  pRpcMsg->contLen = pMsg->bytes;
M
Minghao Li 已提交
429
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
M
Minghao Li 已提交
430
  syncPingSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
M
Minghao Li 已提交
431 432
}

M
Minghao Li 已提交
433 434
void syncPingFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPing* pMsg) {
  syncPingDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
M
Minghao Li 已提交
435 436
}

M
Minghao Li 已提交
437 438 439 440 441
SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg) {
  SyncPing* pMsg = syncPingDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
  return pMsg;
}

M
Minghao Li 已提交
442
cJSON* syncPing2Json(const SyncPing* pMsg) {
M
Minghao Li 已提交
443
  char   u64buf[128];
M
Minghao Li 已提交
444
  cJSON* pRoot = cJSON_CreateObject();
M
Minghao Li 已提交
445 446 447

  if (pMsg != NULL) {
    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
M
Minghao Li 已提交
448
    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
M
Minghao Li 已提交
449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484
    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);

    cJSON* pSrcId = cJSON_CreateObject();
    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
    {
      uint64_t u64 = pMsg->srcId.addr;
      cJSON*   pTmp = pSrcId;
      char     host[128];
      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), "%lu", pMsg->destId.addr);
    cJSON_AddStringToObject(pDestId, "addr", u64buf);
    {
      uint64_t u64 = pMsg->destId.addr;
      cJSON*   pTmp = pDestId;
      char     host[128];
      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);

    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
    char* s;
    s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
    cJSON_AddStringToObject(pRoot, "data", s);
wafwerar's avatar
wafwerar 已提交
485
    taosMemoryFree(s);
M
Minghao Li 已提交
486 487
    s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
    cJSON_AddStringToObject(pRoot, "data2", s);
wafwerar's avatar
wafwerar 已提交
488
    taosMemoryFree(s);
M
Minghao Li 已提交
489
  }
M
Minghao Li 已提交
490

M
Minghao Li 已提交
491 492 493 494 495
  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SyncPing", pRoot);
  return pJson;
}

M
Minghao Li 已提交
496 497 498 499 500
char* syncPing2Str(const SyncPing* pMsg) {
  cJSON* pJson = syncPing2Json(pMsg);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
M
Minghao Li 已提交
501 502
}

M
Minghao Li 已提交
503 504 505 506 507
// for debug ----------------------
void syncPingPrint(const SyncPing* pMsg) {
  char* serialized = syncPing2Str(pMsg);
  printf("syncPingPrint | len:%lu | %s \n", strlen(serialized), serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
508
  taosMemoryFree(serialized);
M
Minghao Li 已提交
509 510 511 512 513 514
}

void syncPingPrint2(char* s, const SyncPing* pMsg) {
  char* serialized = syncPing2Str(pMsg);
  printf("syncPingPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
515
  taosMemoryFree(serialized);
M
Minghao Li 已提交
516 517 518 519 520
}

void syncPingLog(const SyncPing* pMsg) {
  char* serialized = syncPing2Str(pMsg);
  sTrace("syncPingLog | len:%lu | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
521
  taosMemoryFree(serialized);
M
Minghao Li 已提交
522 523 524 525 526
}

void syncPingLog2(char* s, const SyncPing* pMsg) {
  char* serialized = syncPing2Str(pMsg);
  sTrace("syncPingLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
wafwerar's avatar
wafwerar 已提交
527
  taosMemoryFree(serialized);
M
Minghao Li 已提交
528 529
}

M
Minghao Li 已提交
530 531
// ---- message process SyncPingReply----
SyncPingReply* syncPingReplyBuild(uint32_t dataLen) {
M
Minghao Li 已提交
532
  uint32_t       bytes = sizeof(SyncPingReply) + dataLen;
wafwerar's avatar
wafwerar 已提交
533
  SyncPingReply* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
534 535
  memset(pMsg, 0, bytes);
  pMsg->bytes = bytes;
M
Minghao Li 已提交
536
  pMsg->msgType = TDMT_VND_SYNC_PING_REPLY;
M
Minghao Li 已提交
537
  pMsg->dataLen = dataLen;
M
Minghao Li 已提交
538
  return pMsg;
M
Minghao Li 已提交
539 540
}

M
Minghao Li 已提交
541
SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, int32_t vgId, const char* str) {
M
Minghao Li 已提交
542 543
  uint32_t       dataLen = strlen(str) + 1;
  SyncPingReply* pMsg = syncPingReplyBuild(dataLen);
M
Minghao Li 已提交
544
  pMsg->vgId = vgId;
M
Minghao Li 已提交
545 546 547 548 549 550
  pMsg->srcId = *srcId;
  pMsg->destId = *destId;
  snprintf(pMsg->data, pMsg->dataLen, "%s", str);
  return pMsg;
}

M
Minghao Li 已提交
551 552
SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId, int32_t vgId) {
  SyncPingReply* pMsg = syncPingReplyBuild2(srcId, destId, vgId, "pang");
M
Minghao Li 已提交
553 554 555
  return pMsg;
}

M
Minghao Li 已提交
556 557
void syncPingReplyDestroy(SyncPingReply* pMsg) {
  if (pMsg != NULL) {
wafwerar's avatar
wafwerar 已提交
558
    taosMemoryFree(pMsg);
M
Minghao Li 已提交
559 560 561 562 563 564 565 566 567 568 569
  }
}

void syncPingReplySerialize(const SyncPingReply* pMsg, char* buf, uint32_t bufLen) {
  assert(pMsg->bytes <= bufLen);
  memcpy(buf, pMsg, pMsg->bytes);
}

void syncPingReplyDeserialize(const char* buf, uint32_t len, SyncPingReply* pMsg) {
  memcpy(pMsg, buf, len);
  assert(len == pMsg->bytes);
M
Minghao Li 已提交
570
  assert(pMsg->bytes == sizeof(SyncPing) + pMsg->dataLen);
M
Minghao Li 已提交
571 572
}

M
Minghao Li 已提交
573
char* syncPingReplySerialize2(const SyncPingReply* pMsg, uint32_t* len) {
wafwerar's avatar
wafwerar 已提交
574
  char* buf = taosMemoryMalloc(pMsg->bytes);
M
Minghao Li 已提交
575 576 577 578 579 580 581 582 583 584
  assert(buf != NULL);
  syncPingReplySerialize(pMsg, buf, pMsg->bytes);
  if (len != NULL) {
    *len = pMsg->bytes;
  }
  return buf;
}

SyncPingReply* syncPingReplyDeserialize2(const char* buf, uint32_t len) {
  uint32_t       bytes = *((uint32_t*)buf);
wafwerar's avatar
wafwerar 已提交
585
  SyncPingReply* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
586 587 588 589 590 591
  assert(pMsg != NULL);
  syncPingReplyDeserialize(buf, len, pMsg);
  assert(len == pMsg->bytes);
  return pMsg;
}

M
Minghao Li 已提交
592
int32_t syncPingReplySerialize3(const SyncPingReply* pMsg, char* buf, int32_t bufLen) {
H
Hongze Cheng 已提交
593 594
  SEncoder encoder = {0};
  tEncoderInit(&encoder, buf, bufLen);
M
Minghao Li 已提交
595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628
  if (tStartEncode(&encoder) < 0) {
    return -1;
  }

  if (tEncodeU32(&encoder, pMsg->bytes) < 0) {
    return -1;
  }
  if (tEncodeI32(&encoder, pMsg->vgId) < 0) {
    return -1;
  }
  if (tEncodeU32(&encoder, pMsg->msgType) < 0) {
    return -1;
  }
  if (tEncodeU64(&encoder, pMsg->srcId.addr) < 0) {
    return -1;
  }
  if (tEncodeI32(&encoder, pMsg->srcId.vgId) < 0) {
    return -1;
  }
  if (tEncodeU64(&encoder, pMsg->destId.addr) < 0) {
    return -1;
  }
  if (tEncodeI32(&encoder, pMsg->destId.vgId) < 0) {
    return -1;
  }
  if (tEncodeU32(&encoder, pMsg->dataLen) < 0) {
    return -1;
  }
  if (tEncodeBinary(&encoder, pMsg->data, pMsg->dataLen)) {
    return -1;
  }

  tEndEncode(&encoder);
  int32_t tlen = encoder.pos;
H
Hongze Cheng 已提交
629
  tEncoderClear(&encoder);
M
Minghao Li 已提交
630 631 632 633
  return tlen;
}

SyncPingReply* syncPingReplyDeserialize3(void* buf, int32_t bufLen) {
H
Hongze Cheng 已提交
634 635
  SDecoder decoder = {0};
  tDecoderInit(&decoder, buf, bufLen);
M
Minghao Li 已提交
636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670
  if (tStartDecode(&decoder) < 0) {
    return NULL;
  }

  SyncPingReply* pMsg = NULL;
  uint32_t       bytes;
  if (tDecodeU32(&decoder, &bytes) < 0) {
    return NULL;
  }

  pMsg = taosMemoryMalloc(bytes);
  assert(pMsg != NULL);
  pMsg->bytes = bytes;

  if (tDecodeI32(&decoder, &pMsg->vgId) < 0) {
    return NULL;
  }
  if (tDecodeU32(&decoder, &pMsg->msgType) < 0) {
    return NULL;
  }
  if (tDecodeU64(&decoder, &pMsg->srcId.addr) < 0) {
    return NULL;
  }
  if (tDecodeI32(&decoder, &pMsg->srcId.vgId) < 0) {
    return NULL;
  }
  if (tDecodeU64(&decoder, &pMsg->destId.addr) < 0) {
    return NULL;
  }
  if (tDecodeI32(&decoder, &pMsg->destId.vgId) < 0) {
    return NULL;
  }
  if (tDecodeU32(&decoder, &pMsg->dataLen) < 0) {
    return NULL;
  }
H
Hongze Cheng 已提交
671
  uint32_t len;
M
Minghao Li 已提交
672
  char*    data = NULL;
H
Hongze Cheng 已提交
673
  if (tDecodeBinary(&decoder, (uint8_t**)(&data), &len) < 0) {
M
Minghao Li 已提交
674 675 676 677 678 679
    return NULL;
  }
  assert(len = pMsg->dataLen);
  memcpy(pMsg->data, data, len);

  tEndDecode(&decoder);
H
Hongze Cheng 已提交
680
  tDecoderClear(&decoder);
M
Minghao Li 已提交
681 682 683
  return pMsg;
}

M
Minghao Li 已提交
684
void syncPingReply2RpcMsg(const SyncPingReply* pMsg, SRpcMsg* pRpcMsg) {
M
Minghao Li 已提交
685
  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
M
Minghao Li 已提交
686 687 688 689 690 691 692 693 694 695
  pRpcMsg->msgType = pMsg->msgType;
  pRpcMsg->contLen = pMsg->bytes;
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
  syncPingReplySerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
}

void syncPingReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPingReply* pMsg) {
  syncPingReplyDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}

M
Minghao Li 已提交
696 697 698 699 700
SyncPingReply* syncPingReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) {
  SyncPingReply* pMsg = syncPingReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
  return pMsg;
}

M
Minghao Li 已提交
701
cJSON* syncPingReply2Json(const SyncPingReply* pMsg) {
M
Minghao Li 已提交
702
  char   u64buf[128];
M
Minghao Li 已提交
703
  cJSON* pRoot = cJSON_CreateObject();
M
Minghao Li 已提交
704 705 706

  if (pMsg != NULL) {
    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
M
Minghao Li 已提交
707
    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
M
Minghao Li 已提交
708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743
    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);

    cJSON* pSrcId = cJSON_CreateObject();
    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
    {
      uint64_t u64 = pMsg->srcId.addr;
      cJSON*   pTmp = pSrcId;
      char     host[128];
      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), "%lu", pMsg->destId.addr);
    cJSON_AddStringToObject(pDestId, "addr", u64buf);
    {
      uint64_t u64 = pMsg->destId.addr;
      cJSON*   pTmp = pDestId;
      char     host[128];
      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);

    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
    char* s;
    s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
    cJSON_AddStringToObject(pRoot, "data", s);
wafwerar's avatar
wafwerar 已提交
744
    taosMemoryFree(s);
M
Minghao Li 已提交
745 746
    s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
    cJSON_AddStringToObject(pRoot, "data2", s);
wafwerar's avatar
wafwerar 已提交
747
    taosMemoryFree(s);
M
Minghao Li 已提交
748
  }
M
Minghao Li 已提交
749 750 751 752

  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SyncPingReply", pRoot);
  return pJson;
M
Minghao Li 已提交
753 754
}

M
Minghao Li 已提交
755 756 757 758 759
char* syncPingReply2Str(const SyncPingReply* pMsg) {
  cJSON* pJson = syncPingReply2Json(pMsg);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
M
Minghao Li 已提交
760 761
}

M
Minghao Li 已提交
762 763 764
// for debug ----------------------
void syncPingReplyPrint(const SyncPingReply* pMsg) {
  char* serialized = syncPingReply2Str(pMsg);
765
  printf("syncPingReplyPrint | len:%zu | %s \n", strlen(serialized), serialized);
M
Minghao Li 已提交
766
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
767
  taosMemoryFree(serialized);
M
Minghao Li 已提交
768 769 770 771
}

void syncPingReplyPrint2(char* s, const SyncPingReply* pMsg) {
  char* serialized = syncPingReply2Str(pMsg);
772
  printf("syncPingReplyPrint2 | len:%zu | %s | %s \n", strlen(serialized), s, serialized);
M
Minghao Li 已提交
773
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
774
  taosMemoryFree(serialized);
M
Minghao Li 已提交
775 776 777 778
}

void syncPingReplyLog(const SyncPingReply* pMsg) {
  char* serialized = syncPingReply2Str(pMsg);
779
  sTrace("syncPingReplyLog | len:%zu | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
780
  taosMemoryFree(serialized);
M
Minghao Li 已提交
781 782 783 784
}

void syncPingReplyLog2(char* s, const SyncPingReply* pMsg) {
  char* serialized = syncPingReply2Str(pMsg);
785
  sTrace("syncPingReplyLog2 | len:%zu | %s | %s", strlen(serialized), s, serialized);
wafwerar's avatar
wafwerar 已提交
786
  taosMemoryFree(serialized);
M
Minghao Li 已提交
787 788
}

M
Minghao Li 已提交
789 790
// ---- message process SyncClientRequest----
SyncClientRequest* syncClientRequestBuild(uint32_t dataLen) {
M
Minghao Li 已提交
791
  uint32_t           bytes = sizeof(SyncClientRequest) + dataLen;
wafwerar's avatar
wafwerar 已提交
792
  SyncClientRequest* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
793 794
  memset(pMsg, 0, bytes);
  pMsg->bytes = bytes;
M
Minghao Li 已提交
795
  pMsg->msgType = TDMT_VND_SYNC_CLIENT_REQUEST;
M
Minghao Li 已提交
796 797 798 799 800 801
  pMsg->seqNum = 0;
  pMsg->isWeak = false;
  pMsg->dataLen = dataLen;
  return pMsg;
}

M
Minghao Li 已提交
802
// step 1. original SRpcMsg => SyncClientRequest, add seqNum, isWeak
M
Minghao Li 已提交
803
SyncClientRequest* syncClientRequestBuild2(const SRpcMsg* pOriginalRpcMsg, uint64_t seqNum, bool isWeak, int32_t vgId) {
M
Minghao Li 已提交
804
  SyncClientRequest* pMsg = syncClientRequestBuild(pOriginalRpcMsg->contLen);
M
Minghao Li 已提交
805
  pMsg->vgId = vgId;
M
Minghao Li 已提交
806 807 808 809 810 811 812
  pMsg->originalRpcType = pOriginalRpcMsg->msgType;
  pMsg->seqNum = seqNum;
  pMsg->isWeak = isWeak;
  memcpy(pMsg->data, pOriginalRpcMsg->pCont, pOriginalRpcMsg->contLen);
  return pMsg;
}

M
Minghao Li 已提交
813 814
void syncClientRequestDestroy(SyncClientRequest* pMsg) {
  if (pMsg != NULL) {
wafwerar's avatar
wafwerar 已提交
815
    taosMemoryFree(pMsg);
M
Minghao Li 已提交
816 817 818 819 820 821 822 823 824 825 826 827 828
  }
}

void syncClientRequestSerialize(const SyncClientRequest* pMsg, char* buf, uint32_t bufLen) {
  assert(pMsg->bytes <= bufLen);
  memcpy(buf, pMsg, pMsg->bytes);
}

void syncClientRequestDeserialize(const char* buf, uint32_t len, SyncClientRequest* pMsg) {
  memcpy(pMsg, buf, len);
  assert(len == pMsg->bytes);
}

M
Minghao Li 已提交
829
char* syncClientRequestSerialize2(const SyncClientRequest* pMsg, uint32_t* len) {
wafwerar's avatar
wafwerar 已提交
830
  char* buf = taosMemoryMalloc(pMsg->bytes);
M
Minghao Li 已提交
831 832 833 834 835 836 837 838 839 840
  assert(buf != NULL);
  syncClientRequestSerialize(pMsg, buf, pMsg->bytes);
  if (len != NULL) {
    *len = pMsg->bytes;
  }
  return buf;
}

SyncClientRequest* syncClientRequestDeserialize2(const char* buf, uint32_t len) {
  uint32_t           bytes = *((uint32_t*)buf);
wafwerar's avatar
wafwerar 已提交
841
  SyncClientRequest* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
842 843 844 845 846 847
  assert(pMsg != NULL);
  syncClientRequestDeserialize(buf, len, pMsg);
  assert(len == pMsg->bytes);
  return pMsg;
}

M
Minghao Li 已提交
848
// step 2. SyncClientRequest => RpcMsg, to queue
M
Minghao Li 已提交
849 850 851 852 853 854 855 856 857 858 859 860
void syncClientRequest2RpcMsg(const SyncClientRequest* pMsg, SRpcMsg* pRpcMsg) {
  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
  pRpcMsg->msgType = pMsg->msgType;
  pRpcMsg->contLen = pMsg->bytes;
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
  syncClientRequestSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
}

void syncClientRequestFromRpcMsg(const SRpcMsg* pRpcMsg, SyncClientRequest* pMsg) {
  syncClientRequestDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}

M
Minghao Li 已提交
861
// step 3. RpcMsg => SyncClientRequest, from queue
M
Minghao Li 已提交
862 863 864 865 866
SyncClientRequest* syncClientRequestFromRpcMsg2(const SRpcMsg* pRpcMsg) {
  SyncClientRequest* pMsg = syncClientRequestDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
  return pMsg;
}

M
Minghao Li 已提交
867
cJSON* syncClientRequest2Json(const SyncClientRequest* pMsg) {
M
Minghao Li 已提交
868
  char   u64buf[128];
M
Minghao Li 已提交
869
  cJSON* pRoot = cJSON_CreateObject();
M
Minghao Li 已提交
870 871 872

  if (pMsg != NULL) {
    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
M
Minghao Li 已提交
873
    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
M
Minghao Li 已提交
874 875 876 877 878 879 880 881 882 883
    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
    cJSON_AddNumberToObject(pRoot, "originalRpcType", pMsg->originalRpcType);
    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->seqNum);
    cJSON_AddStringToObject(pRoot, "seqNum", u64buf);
    cJSON_AddNumberToObject(pRoot, "isWeak", pMsg->isWeak);
    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);

    char* s;
    s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
    cJSON_AddStringToObject(pRoot, "data", s);
wafwerar's avatar
wafwerar 已提交
884
    taosMemoryFree(s);
M
Minghao Li 已提交
885 886
    s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
    cJSON_AddStringToObject(pRoot, "data2", s);
wafwerar's avatar
wafwerar 已提交
887
    taosMemoryFree(s);
M
Minghao Li 已提交
888
  }
M
Minghao Li 已提交
889

M
Minghao Li 已提交
890 891 892 893 894
  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SyncClientRequest", pRoot);
  return pJson;
}

M
Minghao Li 已提交
895 896 897 898 899 900 901 902 903 904 905 906
char* syncClientRequest2Str(const SyncClientRequest* pMsg) {
  cJSON* pJson = syncClientRequest2Json(pMsg);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}

// for debug ----------------------
void syncClientRequestPrint(const SyncClientRequest* pMsg) {
  char* serialized = syncClientRequest2Str(pMsg);
  printf("syncClientRequestPrint | len:%lu | %s \n", strlen(serialized), serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
907
  taosMemoryFree(serialized);
M
Minghao Li 已提交
908 909 910 911 912 913
}

void syncClientRequestPrint2(char* s, const SyncClientRequest* pMsg) {
  char* serialized = syncClientRequest2Str(pMsg);
  printf("syncClientRequestPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
914
  taosMemoryFree(serialized);
M
Minghao Li 已提交
915 916 917 918 919
}

void syncClientRequestLog(const SyncClientRequest* pMsg) {
  char* serialized = syncClientRequest2Str(pMsg);
  sTrace("syncClientRequestLog | len:%lu | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
920
  taosMemoryFree(serialized);
M
Minghao Li 已提交
921 922 923 924 925
}

void syncClientRequestLog2(char* s, const SyncClientRequest* pMsg) {
  char* serialized = syncClientRequest2Str(pMsg);
  sTrace("syncClientRequestLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
wafwerar's avatar
wafwerar 已提交
926
  taosMemoryFree(serialized);
M
Minghao Li 已提交
927 928
}

M
Minghao Li 已提交
929
// ---- message process SyncRequestVote----
M
Minghao Li 已提交
930
SyncRequestVote* syncRequestVoteBuild(int32_t vgId) {
M
Minghao Li 已提交
931
  uint32_t         bytes = sizeof(SyncRequestVote);
wafwerar's avatar
wafwerar 已提交
932
  SyncRequestVote* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
933 934
  memset(pMsg, 0, bytes);
  pMsg->bytes = bytes;
M
Minghao Li 已提交
935 936
  pMsg->vgId = vgId;
  pMsg->msgType = TDMT_VND_SYNC_REQUEST_VOTE;
M
Minghao Li 已提交
937
  return pMsg;
M
Minghao Li 已提交
938 939
}

M
Minghao Li 已提交
940 941
void syncRequestVoteDestroy(SyncRequestVote* pMsg) {
  if (pMsg != NULL) {
wafwerar's avatar
wafwerar 已提交
942
    taosMemoryFree(pMsg);
M
Minghao Li 已提交
943 944
  }
}
M
Minghao Li 已提交
945

M
Minghao Li 已提交
946 947 948 949
void syncRequestVoteSerialize(const SyncRequestVote* pMsg, char* buf, uint32_t bufLen) {
  assert(pMsg->bytes <= bufLen);
  memcpy(buf, pMsg, pMsg->bytes);
}
M
Minghao Li 已提交
950

M
Minghao Li 已提交
951 952 953 954
void syncRequestVoteDeserialize(const char* buf, uint32_t len, SyncRequestVote* pMsg) {
  memcpy(pMsg, buf, len);
  assert(len == pMsg->bytes);
}
M
Minghao Li 已提交
955

M
Minghao Li 已提交
956
char* syncRequestVoteSerialize2(const SyncRequestVote* pMsg, uint32_t* len) {
wafwerar's avatar
wafwerar 已提交
957
  char* buf = taosMemoryMalloc(pMsg->bytes);
M
Minghao Li 已提交
958 959 960 961 962 963 964 965 966 967
  assert(buf != NULL);
  syncRequestVoteSerialize(pMsg, buf, pMsg->bytes);
  if (len != NULL) {
    *len = pMsg->bytes;
  }
  return buf;
}

SyncRequestVote* syncRequestVoteDeserialize2(const char* buf, uint32_t len) {
  uint32_t         bytes = *((uint32_t*)buf);
wafwerar's avatar
wafwerar 已提交
968
  SyncRequestVote* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
969 970 971 972 973 974
  assert(pMsg != NULL);
  syncRequestVoteDeserialize(buf, len, pMsg);
  assert(len == pMsg->bytes);
  return pMsg;
}

M
Minghao Li 已提交
975 976 977 978 979 980 981
void syncRequestVote2RpcMsg(const SyncRequestVote* pMsg, SRpcMsg* pRpcMsg) {
  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
  pRpcMsg->msgType = pMsg->msgType;
  pRpcMsg->contLen = pMsg->bytes;
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
  syncRequestVoteSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
}
M
Minghao Li 已提交
982

M
Minghao Li 已提交
983 984 985
void syncRequestVoteFromRpcMsg(const SRpcMsg* pRpcMsg, SyncRequestVote* pMsg) {
  syncRequestVoteDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}
M
Minghao Li 已提交
986

M
Minghao Li 已提交
987 988
SyncRequestVote* syncRequestVoteFromRpcMsg2(const SRpcMsg* pRpcMsg) {
  SyncRequestVote* pMsg = syncRequestVoteDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
M
Minghao Li 已提交
989
  return pMsg;
M
Minghao Li 已提交
990 991
}

M
Minghao Li 已提交
992
cJSON* syncRequestVote2Json(const SyncRequestVote* pMsg) {
M
Minghao Li 已提交
993
  char   u64buf[128];
M
Minghao Li 已提交
994 995
  cJSON* pRoot = cJSON_CreateObject();

M
Minghao Li 已提交
996 997
  if (pMsg != NULL) {
    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
M
Minghao Li 已提交
998
    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
M
Minghao Li 已提交
999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031
    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);

    cJSON* pSrcId = cJSON_CreateObject();
    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
    {
      uint64_t u64 = pMsg->srcId.addr;
      cJSON*   pTmp = pSrcId;
      char     host[128];
      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();
    cJSON_AddNumberToObject(pDestId, "addr", pMsg->destId.addr);
    {
      uint64_t u64 = pMsg->destId.addr;
      cJSON*   pTmp = pDestId;
      char     host[128];
      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), "%lu", pMsg->term);
    cJSON_AddStringToObject(pRoot, "term", u64buf);
M
Minghao Li 已提交
1032
    snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->lastLogIndex);
M
Minghao Li 已提交
1033 1034 1035 1036
    cJSON_AddStringToObject(pRoot, "lastLogIndex", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->lastLogTerm);
    cJSON_AddStringToObject(pRoot, "lastLogTerm", u64buf);
  }
M
Minghao Li 已提交
1037 1038 1039 1040

  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SyncRequestVote", pRoot);
  return pJson;
M
Minghao Li 已提交
1041
}
M
Minghao Li 已提交
1042

M
Minghao Li 已提交
1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054
char* syncRequestVote2Str(const SyncRequestVote* pMsg) {
  cJSON* pJson = syncRequestVote2Json(pMsg);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}

// for debug ----------------------
void syncRequestVotePrint(const SyncRequestVote* pMsg) {
  char* serialized = syncRequestVote2Str(pMsg);
  printf("syncRequestVotePrint | len:%lu | %s \n", strlen(serialized), serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
1055
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1056 1057 1058 1059 1060 1061
}

void syncRequestVotePrint2(char* s, const SyncRequestVote* pMsg) {
  char* serialized = syncRequestVote2Str(pMsg);
  printf("syncRequestVotePrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
1062
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1063 1064 1065 1066 1067
}

void syncRequestVoteLog(const SyncRequestVote* pMsg) {
  char* serialized = syncRequestVote2Str(pMsg);
  sTrace("syncRequestVoteLog | len:%lu | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
1068
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1069 1070 1071 1072 1073
}

void syncRequestVoteLog2(char* s, const SyncRequestVote* pMsg) {
  char* serialized = syncRequestVote2Str(pMsg);
  sTrace("syncRequestVoteLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
wafwerar's avatar
wafwerar 已提交
1074
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1075 1076
}

M
Minghao Li 已提交
1077
// ---- message process SyncRequestVoteReply----
M
Minghao Li 已提交
1078
SyncRequestVoteReply* syncRequestVoteReplyBuild(int32_t vgId) {
M
Minghao Li 已提交
1079
  uint32_t              bytes = sizeof(SyncRequestVoteReply);
wafwerar's avatar
wafwerar 已提交
1080
  SyncRequestVoteReply* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
1081 1082
  memset(pMsg, 0, bytes);
  pMsg->bytes = bytes;
M
Minghao Li 已提交
1083 1084
  pMsg->vgId = vgId;
  pMsg->msgType = TDMT_VND_SYNC_REQUEST_VOTE_REPLY;
M
Minghao Li 已提交
1085
  return pMsg;
M
Minghao Li 已提交
1086 1087 1088 1089
}

void syncRequestVoteReplyDestroy(SyncRequestVoteReply* pMsg) {
  if (pMsg != NULL) {
wafwerar's avatar
wafwerar 已提交
1090
    taosMemoryFree(pMsg);
M
Minghao Li 已提交
1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103
  }
}

void syncRequestVoteReplySerialize(const SyncRequestVoteReply* pMsg, char* buf, uint32_t bufLen) {
  assert(pMsg->bytes <= bufLen);
  memcpy(buf, pMsg, pMsg->bytes);
}

void syncRequestVoteReplyDeserialize(const char* buf, uint32_t len, SyncRequestVoteReply* pMsg) {
  memcpy(pMsg, buf, len);
  assert(len == pMsg->bytes);
}

M
Minghao Li 已提交
1104
char* syncRequestVoteReplySerialize2(const SyncRequestVoteReply* pMsg, uint32_t* len) {
wafwerar's avatar
wafwerar 已提交
1105
  char* buf = taosMemoryMalloc(pMsg->bytes);
M
Minghao Li 已提交
1106 1107 1108 1109 1110 1111 1112 1113 1114 1115
  assert(buf != NULL);
  syncRequestVoteReplySerialize(pMsg, buf, pMsg->bytes);
  if (len != NULL) {
    *len = pMsg->bytes;
  }
  return buf;
}

SyncRequestVoteReply* syncRequestVoteReplyDeserialize2(const char* buf, uint32_t len) {
  uint32_t              bytes = *((uint32_t*)buf);
wafwerar's avatar
wafwerar 已提交
1116
  SyncRequestVoteReply* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
1117 1118 1119 1120 1121 1122
  assert(pMsg != NULL);
  syncRequestVoteReplyDeserialize(buf, len, pMsg);
  assert(len == pMsg->bytes);
  return pMsg;
}

M
Minghao Li 已提交
1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134
void syncRequestVoteReply2RpcMsg(const SyncRequestVoteReply* pMsg, SRpcMsg* pRpcMsg) {
  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
  pRpcMsg->msgType = pMsg->msgType;
  pRpcMsg->contLen = pMsg->bytes;
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
  syncRequestVoteReplySerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
}

void syncRequestVoteReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncRequestVoteReply* pMsg) {
  syncRequestVoteReplyDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}

M
Minghao Li 已提交
1135 1136 1137 1138 1139
SyncRequestVoteReply* syncRequestVoteReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) {
  SyncRequestVoteReply* pMsg = syncRequestVoteReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
  return pMsg;
}

M
Minghao Li 已提交
1140
cJSON* syncRequestVoteReply2Json(const SyncRequestVoteReply* pMsg) {
M
Minghao Li 已提交
1141
  char   u64buf[128];
M
Minghao Li 已提交
1142 1143
  cJSON* pRoot = cJSON_CreateObject();

M
Minghao Li 已提交
1144 1145
  if (pMsg != NULL) {
    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
M
Minghao Li 已提交
1146
    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
M
Minghao Li 已提交
1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181
    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);

    cJSON* pSrcId = cJSON_CreateObject();
    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
    {
      uint64_t u64 = pMsg->srcId.addr;
      cJSON*   pTmp = pSrcId;
      char     host[128];
      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();
    cJSON_AddNumberToObject(pDestId, "addr", pMsg->destId.addr);
    {
      uint64_t u64 = pMsg->destId.addr;
      cJSON*   pTmp = pDestId;
      char     host[128];
      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), "%lu", pMsg->term);
    cJSON_AddStringToObject(pRoot, "term", u64buf);
    cJSON_AddNumberToObject(pRoot, "vote_granted", pMsg->voteGranted);
  }
M
Minghao Li 已提交
1182 1183 1184 1185

  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SyncRequestVoteReply", pRoot);
  return pJson;
M
Minghao Li 已提交
1186 1187
}

M
Minghao Li 已提交
1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199
char* syncRequestVoteReply2Str(const SyncRequestVoteReply* pMsg) {
  cJSON* pJson = syncRequestVoteReply2Json(pMsg);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}

// for debug ----------------------
void syncRequestVoteReplyPrint(const SyncRequestVoteReply* pMsg) {
  char* serialized = syncRequestVoteReply2Str(pMsg);
  printf("syncRequestVoteReplyPrint | len:%lu | %s \n", strlen(serialized), serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
1200
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1201 1202 1203 1204 1205 1206
}

void syncRequestVoteReplyPrint2(char* s, const SyncRequestVoteReply* pMsg) {
  char* serialized = syncRequestVoteReply2Str(pMsg);
  printf("syncRequestVoteReplyPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
1207
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1208 1209 1210 1211 1212
}

void syncRequestVoteReplyLog(const SyncRequestVoteReply* pMsg) {
  char* serialized = syncRequestVoteReply2Str(pMsg);
  sTrace("syncRequestVoteReplyLog | len:%lu | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
1213
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1214 1215 1216 1217 1218
}

void syncRequestVoteReplyLog2(char* s, const SyncRequestVoteReply* pMsg) {
  char* serialized = syncRequestVoteReply2Str(pMsg);
  sTrace("syncRequestVoteReplyLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
wafwerar's avatar
wafwerar 已提交
1219
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1220 1221
}

M
Minghao Li 已提交
1222
// ---- message process SyncAppendEntries----
M
Minghao Li 已提交
1223
SyncAppendEntries* syncAppendEntriesBuild(uint32_t dataLen, int32_t vgId) {
M
Minghao Li 已提交
1224
  uint32_t           bytes = sizeof(SyncAppendEntries) + dataLen;
wafwerar's avatar
wafwerar 已提交
1225
  SyncAppendEntries* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
1226 1227
  memset(pMsg, 0, bytes);
  pMsg->bytes = bytes;
M
Minghao Li 已提交
1228 1229
  pMsg->vgId = vgId;
  pMsg->msgType = TDMT_VND_SYNC_APPEND_ENTRIES;
M
Minghao Li 已提交
1230
  pMsg->dataLen = dataLen;
M
Minghao Li 已提交
1231
  return pMsg;
M
Minghao Li 已提交
1232 1233 1234 1235
}

void syncAppendEntriesDestroy(SyncAppendEntries* pMsg) {
  if (pMsg != NULL) {
wafwerar's avatar
wafwerar 已提交
1236
    taosMemoryFree(pMsg);
M
Minghao Li 已提交
1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247
  }
}

void syncAppendEntriesSerialize(const SyncAppendEntries* pMsg, char* buf, uint32_t bufLen) {
  assert(pMsg->bytes <= bufLen);
  memcpy(buf, pMsg, pMsg->bytes);
}

void syncAppendEntriesDeserialize(const char* buf, uint32_t len, SyncAppendEntries* pMsg) {
  memcpy(pMsg, buf, len);
  assert(len == pMsg->bytes);
M
Minghao Li 已提交
1248 1249 1250 1251
  assert(pMsg->bytes == sizeof(SyncAppendEntries) + pMsg->dataLen);
}

char* syncAppendEntriesSerialize2(const SyncAppendEntries* pMsg, uint32_t* len) {
wafwerar's avatar
wafwerar 已提交
1252
  char* buf = taosMemoryMalloc(pMsg->bytes);
M
Minghao Li 已提交
1253 1254 1255 1256 1257 1258 1259 1260 1261 1262
  assert(buf != NULL);
  syncAppendEntriesSerialize(pMsg, buf, pMsg->bytes);
  if (len != NULL) {
    *len = pMsg->bytes;
  }
  return buf;
}

SyncAppendEntries* syncAppendEntriesDeserialize2(const char* buf, uint32_t len) {
  uint32_t           bytes = *((uint32_t*)buf);
wafwerar's avatar
wafwerar 已提交
1263
  SyncAppendEntries* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
1264 1265 1266 1267
  assert(pMsg != NULL);
  syncAppendEntriesDeserialize(buf, len, pMsg);
  assert(len == pMsg->bytes);
  return pMsg;
M
Minghao Li 已提交
1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281
}

void syncAppendEntries2RpcMsg(const SyncAppendEntries* pMsg, SRpcMsg* pRpcMsg) {
  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
  pRpcMsg->msgType = pMsg->msgType;
  pRpcMsg->contLen = pMsg->bytes;
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
  syncAppendEntriesSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
}

void syncAppendEntriesFromRpcMsg(const SRpcMsg* pRpcMsg, SyncAppendEntries* pMsg) {
  syncAppendEntriesDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}

M
Minghao Li 已提交
1282 1283 1284 1285 1286
SyncAppendEntries* syncAppendEntriesFromRpcMsg2(const SRpcMsg* pRpcMsg) {
  SyncAppendEntries* pMsg = syncAppendEntriesDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
  return pMsg;
}

M
Minghao Li 已提交
1287
cJSON* syncAppendEntries2Json(const SyncAppendEntries* pMsg) {
M
Minghao Li 已提交
1288
  char   u64buf[128];
M
Minghao Li 已提交
1289 1290
  cJSON* pRoot = cJSON_CreateObject();

M
Minghao Li 已提交
1291 1292
  if (pMsg != NULL) {
    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
M
Minghao Li 已提交
1293
    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
M
Minghao Li 已提交
1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328
    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);

    cJSON* pSrcId = cJSON_CreateObject();
    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
    {
      uint64_t u64 = pMsg->srcId.addr;
      cJSON*   pTmp = pSrcId;
      char     host[128];
      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), "%lu", pMsg->destId.addr);
    cJSON_AddStringToObject(pDestId, "addr", u64buf);
    {
      uint64_t u64 = pMsg->destId.addr;
      cJSON*   pTmp = pDestId;
      char     host[128];
      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), "%lu", pMsg->term);
    cJSON_AddStringToObject(pRoot, "term", u64buf);

M
Minghao Li 已提交
1329 1330
    snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->prevLogIndex);
    cJSON_AddStringToObject(pRoot, "prevLogIndex", u64buf);
M
Minghao Li 已提交
1331 1332 1333 1334

    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->prevLogTerm);
    cJSON_AddStringToObject(pRoot, "pre_log_term", u64buf);

M
Minghao Li 已提交
1335 1336
    snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->commitIndex);
    cJSON_AddStringToObject(pRoot, "commitIndex", u64buf);
M
Minghao Li 已提交
1337 1338 1339 1340 1341

    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
    char* s;
    s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
    cJSON_AddStringToObject(pRoot, "data", s);
wafwerar's avatar
wafwerar 已提交
1342
    taosMemoryFree(s);
M
Minghao Li 已提交
1343 1344
    s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
    cJSON_AddStringToObject(pRoot, "data2", s);
wafwerar's avatar
wafwerar 已提交
1345
    taosMemoryFree(s);
M
Minghao Li 已提交
1346
  }
M
Minghao Li 已提交
1347 1348 1349 1350 1351 1352

  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SyncAppendEntries", pRoot);
  return pJson;
}

M
Minghao Li 已提交
1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364
char* syncAppendEntries2Str(const SyncAppendEntries* pMsg) {
  cJSON* pJson = syncAppendEntries2Json(pMsg);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}

// for debug ----------------------
void syncAppendEntriesPrint(const SyncAppendEntries* pMsg) {
  char* serialized = syncAppendEntries2Str(pMsg);
  printf("syncAppendEntriesPrint | len:%lu | %s \n", strlen(serialized), serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
1365
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1366 1367 1368 1369 1370 1371
}

void syncAppendEntriesPrint2(char* s, const SyncAppendEntries* pMsg) {
  char* serialized = syncAppendEntries2Str(pMsg);
  printf("syncAppendEntriesPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
1372
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1373 1374 1375 1376 1377
}

void syncAppendEntriesLog(const SyncAppendEntries* pMsg) {
  char* serialized = syncAppendEntries2Str(pMsg);
  sTrace("syncAppendEntriesLog | len:%lu | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
1378
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1379 1380 1381 1382 1383
}

void syncAppendEntriesLog2(char* s, const SyncAppendEntries* pMsg) {
  char* serialized = syncAppendEntries2Str(pMsg);
  sTrace("syncAppendEntriesLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
wafwerar's avatar
wafwerar 已提交
1384
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1385 1386
}

M
Minghao Li 已提交
1387
// ---- message process SyncAppendEntriesReply----
M
Minghao Li 已提交
1388
SyncAppendEntriesReply* syncAppendEntriesReplyBuild(int32_t vgId) {
M
Minghao Li 已提交
1389
  uint32_t                bytes = sizeof(SyncAppendEntriesReply);
wafwerar's avatar
wafwerar 已提交
1390
  SyncAppendEntriesReply* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
1391 1392
  memset(pMsg, 0, bytes);
  pMsg->bytes = bytes;
M
Minghao Li 已提交
1393 1394
  pMsg->vgId = vgId;
  pMsg->msgType = TDMT_VND_SYNC_APPEND_ENTRIES_REPLY;
M
Minghao Li 已提交
1395
  return pMsg;
M
Minghao Li 已提交
1396 1397 1398 1399
}

void syncAppendEntriesReplyDestroy(SyncAppendEntriesReply* pMsg) {
  if (pMsg != NULL) {
wafwerar's avatar
wafwerar 已提交
1400
    taosMemoryFree(pMsg);
M
Minghao Li 已提交
1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413
  }
}

void syncAppendEntriesReplySerialize(const SyncAppendEntriesReply* pMsg, char* buf, uint32_t bufLen) {
  assert(pMsg->bytes <= bufLen);
  memcpy(buf, pMsg, pMsg->bytes);
}

void syncAppendEntriesReplyDeserialize(const char* buf, uint32_t len, SyncAppendEntriesReply* pMsg) {
  memcpy(pMsg, buf, len);
  assert(len == pMsg->bytes);
}

M
Minghao Li 已提交
1414
char* syncAppendEntriesReplySerialize2(const SyncAppendEntriesReply* pMsg, uint32_t* len) {
wafwerar's avatar
wafwerar 已提交
1415
  char* buf = taosMemoryMalloc(pMsg->bytes);
M
Minghao Li 已提交
1416 1417 1418 1419 1420 1421 1422 1423 1424 1425
  assert(buf != NULL);
  syncAppendEntriesReplySerialize(pMsg, buf, pMsg->bytes);
  if (len != NULL) {
    *len = pMsg->bytes;
  }
  return buf;
}

SyncAppendEntriesReply* syncAppendEntriesReplyDeserialize2(const char* buf, uint32_t len) {
  uint32_t                bytes = *((uint32_t*)buf);
wafwerar's avatar
wafwerar 已提交
1426
  SyncAppendEntriesReply* pMsg = taosMemoryMalloc(bytes);
M
Minghao Li 已提交
1427 1428 1429 1430 1431 1432
  assert(pMsg != NULL);
  syncAppendEntriesReplyDeserialize(buf, len, pMsg);
  assert(len == pMsg->bytes);
  return pMsg;
}

M
Minghao Li 已提交
1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444
void syncAppendEntriesReply2RpcMsg(const SyncAppendEntriesReply* pMsg, SRpcMsg* pRpcMsg) {
  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
  pRpcMsg->msgType = pMsg->msgType;
  pRpcMsg->contLen = pMsg->bytes;
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
  syncAppendEntriesReplySerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
}

void syncAppendEntriesReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncAppendEntriesReply* pMsg) {
  syncAppendEntriesReplyDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}

M
Minghao Li 已提交
1445 1446 1447 1448 1449
SyncAppendEntriesReply* syncAppendEntriesReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) {
  SyncAppendEntriesReply* pMsg = syncAppendEntriesReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
  return pMsg;
}

M
Minghao Li 已提交
1450
cJSON* syncAppendEntriesReply2Json(const SyncAppendEntriesReply* pMsg) {
M
Minghao Li 已提交
1451
  char   u64buf[128];
M
Minghao Li 已提交
1452
  cJSON* pRoot = cJSON_CreateObject();
M
Minghao Li 已提交
1453 1454 1455

  if (pMsg != NULL) {
    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
M
Minghao Li 已提交
1456
    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
M
Minghao Li 已提交
1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491
    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);

    cJSON* pSrcId = cJSON_CreateObject();
    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->srcId.addr);
    cJSON_AddStringToObject(pSrcId, "addr", u64buf);
    {
      uint64_t u64 = pMsg->srcId.addr;
      cJSON*   pTmp = pSrcId;
      char     host[128];
      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), "%lu", pMsg->destId.addr);
    cJSON_AddStringToObject(pDestId, "addr", u64buf);
    {
      uint64_t u64 = pMsg->destId.addr;
      cJSON*   pTmp = pDestId;
      char     host[128];
      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), "%lu", pMsg->term);
    cJSON_AddStringToObject(pRoot, "term", u64buf);
    cJSON_AddNumberToObject(pRoot, "success", pMsg->success);
M
Minghao Li 已提交
1492
    snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->matchIndex);
M
Minghao Li 已提交
1493
    cJSON_AddStringToObject(pRoot, "matchIndex", u64buf);
M
Minghao Li 已提交
1494 1495 1496 1497 1498
  }

  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SyncAppendEntriesReply", pRoot);
  return pJson;
M
Minghao Li 已提交
1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512
}

char* syncAppendEntriesReply2Str(const SyncAppendEntriesReply* pMsg) {
  cJSON* pJson = syncAppendEntriesReply2Json(pMsg);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}

// for debug ----------------------
void syncAppendEntriesReplyPrint(const SyncAppendEntriesReply* pMsg) {
  char* serialized = syncAppendEntriesReply2Str(pMsg);
  printf("syncAppendEntriesReplyPrint | len:%lu | %s \n", strlen(serialized), serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
1513
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1514 1515 1516 1517 1518 1519
}

void syncAppendEntriesReplyPrint2(char* s, const SyncAppendEntriesReply* pMsg) {
  char* serialized = syncAppendEntriesReply2Str(pMsg);
  printf("syncAppendEntriesReplyPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
1520
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1521 1522 1523 1524 1525
}

void syncAppendEntriesReplyLog(const SyncAppendEntriesReply* pMsg) {
  char* serialized = syncAppendEntriesReply2Str(pMsg);
  sTrace("syncAppendEntriesReplyLog | len:%lu | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
1526
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1527 1528 1529 1530 1531
}

void syncAppendEntriesReplyLog2(char* s, const SyncAppendEntriesReply* pMsg) {
  char* serialized = syncAppendEntriesReply2Str(pMsg);
  sTrace("syncAppendEntriesReplyLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
wafwerar's avatar
wafwerar 已提交
1532
  taosMemoryFree(serialized);
M
Minghao Li 已提交
1533
}
M
Minghao Li 已提交
1534 1535

// ---- message process SyncApplyMsg----
M
Minghao Li 已提交
1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588
SyncApplyMsg* syncApplyMsgBuild(uint32_t dataLen) {
  uint32_t      bytes = sizeof(SyncApplyMsg) + dataLen;
  SyncApplyMsg* pMsg = taosMemoryMalloc(bytes);
  memset(pMsg, 0, bytes);
  pMsg->bytes = bytes;
  pMsg->msgType = TDMT_VND_SYNC_APPLY_MSG;
  pMsg->dataLen = dataLen;
  return pMsg;
}

SyncApplyMsg* syncApplyMsgBuild2(const SRpcMsg* pOriginalRpcMsg, int32_t vgId, SFsmCbMeta* pMeta) {
  SyncApplyMsg* pMsg = syncApplyMsgBuild(pOriginalRpcMsg->contLen);
  pMsg->vgId = vgId;
  pMsg->originalRpcType = pOriginalRpcMsg->msgType;
  pMsg->fsmMeta = *pMeta;
  memcpy(pMsg->data, pOriginalRpcMsg->pCont, pOriginalRpcMsg->contLen);
  return pMsg;
}

void syncApplyMsgDestroy(SyncApplyMsg* pMsg) {
  if (pMsg != NULL) {
    taosMemoryFree(pMsg);
  }
}

void syncApplyMsgSerialize(const SyncApplyMsg* pMsg, char* buf, uint32_t bufLen) {
  assert(pMsg->bytes <= bufLen);
  memcpy(buf, pMsg, pMsg->bytes);
}

void syncApplyMsgDeserialize(const char* buf, uint32_t len, SyncApplyMsg* pMsg) {
  memcpy(pMsg, buf, len);
  assert(len == pMsg->bytes);
}

char* syncApplyMsgSerialize2(const SyncApplyMsg* pMsg, uint32_t* len) {
  char* buf = taosMemoryMalloc(pMsg->bytes);
  assert(buf != NULL);
  syncApplyMsgSerialize(pMsg, buf, pMsg->bytes);
  if (len != NULL) {
    *len = pMsg->bytes;
  }
  return buf;
}

SyncApplyMsg* syncApplyMsgDeserialize2(const char* buf, uint32_t len) {
  uint32_t      bytes = *((uint32_t*)buf);
  SyncApplyMsg* pMsg = taosMemoryMalloc(bytes);
  assert(pMsg != NULL);
  syncApplyMsgDeserialize(buf, len, pMsg);
  assert(len == pMsg->bytes);
  return pMsg;
}
M
Minghao Li 已提交
1589 1590

// SyncApplyMsg to SRpcMsg, put it into ApplyQ
M
Minghao Li 已提交
1591 1592 1593 1594 1595 1596 1597
void syncApplyMsg2RpcMsg(const SyncApplyMsg* pMsg, SRpcMsg* pRpcMsg) {
  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
  pRpcMsg->msgType = pMsg->msgType;
  pRpcMsg->contLen = pMsg->bytes;
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
  syncApplyMsgSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
}
M
Minghao Li 已提交
1598 1599

// get SRpcMsg from ApplyQ, to SyncApplyMsg
M
Minghao Li 已提交
1600 1601 1602 1603 1604 1605 1606 1607
void syncApplyMsgFromRpcMsg(const SRpcMsg* pRpcMsg, SyncApplyMsg* pMsg) {
  syncApplyMsgDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}

SyncApplyMsg* syncApplyMsgFromRpcMsg2(const SRpcMsg* pRpcMsg) {
  SyncApplyMsg* pMsg = syncApplyMsgDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
  return pMsg;
}
M
Minghao Li 已提交
1608 1609

// SyncApplyMsg to OriginalRpcMsg
M
Minghao Li 已提交
1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645
void syncApplyMsg2OriginalRpcMsg(const SyncApplyMsg* pMsg, SRpcMsg* pOriginalRpcMsg) {
  memset(pOriginalRpcMsg, 0, sizeof(*pOriginalRpcMsg));
  pOriginalRpcMsg->msgType = pMsg->originalRpcType;
  pOriginalRpcMsg->contLen = pMsg->dataLen;
  pOriginalRpcMsg->pCont = rpcMallocCont(pOriginalRpcMsg->contLen);
  memcpy(pOriginalRpcMsg->pCont, pMsg->data, pOriginalRpcMsg->contLen);
}

cJSON* syncApplyMsg2Json(const SyncApplyMsg* pMsg) {
  char   u64buf[128];
  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_AddNumberToObject(pRoot, "originalRpcType", pMsg->originalRpcType);

    snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->fsmMeta.index);
    cJSON_AddStringToObject(pRoot, "fsmMeta.index", u64buf);
    cJSON_AddNumberToObject(pRoot, "fsmMeta.isWeak", pMsg->fsmMeta.isWeak);
    cJSON_AddNumberToObject(pRoot, "fsmMeta.code", pMsg->fsmMeta.code);
    cJSON_AddNumberToObject(pRoot, "fsmMeta.state", pMsg->fsmMeta.state);
    cJSON_AddStringToObject(pRoot, "fsmMeta.state.str", syncUtilState2String(pMsg->fsmMeta.state));
    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->fsmMeta.seqNum);
    cJSON_AddStringToObject(pRoot, "fsmMeta.seqNum", u64buf);

    cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
    char* s;
    s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
    cJSON_AddStringToObject(pRoot, "data", s);
    taosMemoryFree(s);
    s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
    cJSON_AddStringToObject(pRoot, "data2", s);
    taosMemoryFree(s);
  }
M
Minghao Li 已提交
1646

M
Minghao Li 已提交
1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657
  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SyncApplyMsg", pRoot);
  return pJson;
}

char* syncApplyMsg2Str(const SyncApplyMsg* pMsg) {
  cJSON* pJson = syncApplyMsg2Json(pMsg);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}
M
Minghao Li 已提交
1658 1659

// for debug ----------------------
M
Minghao Li 已提交
1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684
void syncApplyMsgPrint(const SyncApplyMsg* pMsg) {
  char* serialized = syncApplyMsg2Str(pMsg);
  printf("syncApplyMsgPrint | len:%lu | %s \n", strlen(serialized), serialized);
  fflush(NULL);
  taosMemoryFree(serialized);
}

void syncApplyMsgPrint2(char* s, const SyncApplyMsg* pMsg) {
  char* serialized = syncApplyMsg2Str(pMsg);
  printf("syncApplyMsgPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
  fflush(NULL);
  taosMemoryFree(serialized);
}

void syncApplyMsgLog(const SyncApplyMsg* pMsg) {
  char* serialized = syncApplyMsg2Str(pMsg);
  sTrace("ssyncApplyMsgLog | len:%lu | %s", strlen(serialized), serialized);
  taosMemoryFree(serialized);
}

void syncApplyMsgLog2(char* s, const SyncApplyMsg* pMsg) {
  char* serialized = syncApplyMsg2Str(pMsg);
  sTrace("syncApplyMsgLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
  taosMemoryFree(serialized);
}