taosdata / TDengine / #5024

coverage: 72.954% (+0.7%) from 72.251%
#5024

push

travis-ci

web-flow 
merge: from main to 3.0 branch #35157

120 of 200 new or added lines in 14 files covered. (60.0%)

655 existing lines in 130 files now uncovered.

273150 of 374416 relevant lines covered (72.95%)

129604715.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.68
/source/dnode/mnode/impl/src/mndDef.c
1 
/*
2 
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3 
 *
4 
 * This program is free software: you can use, redistribute, and/or modify
5 
 * it under the terms of the GNU Affero General Public License, version 3
6 
 * or later ("AGPL"), as published by the Free Software Foundation.
7 
 *
8 
 * This program is distributed in the hope that it will be useful, but WITHOUT
9 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 
 * FITNESS FOR A PARTICULAR PURPOSE.
11 
 *
12 
 * You should have received a copy of the GNU Affero General Public License
13 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14 
 */
15 
#define _DEFAULT_SOURCE
16 
#include "mndConsumer.h"
17 
#include "mndDef.h"
18 
#include "mndStream.h"
19 
#include "taoserror.h"
20 
#include "tunit.h"
21 









22



int32_t tEncodeSStreamObj(SEncoder *pEncoder, const SStreamObj *pObj) {




821,156✔





23



  TAOS_CHECK_RETURN(tStartEncode(pEncoder));




821,156✔





24












25



  TAOS_CHECK_RETURN(tEncodeCStr(pEncoder, pObj->name));




1,642,312✔





26



  TAOS_CHECK_RETURN(tSerializeSCMCreateStreamReqImpl(pEncoder, pObj->pCreate));




821,156✔





27












28



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pObj->mainSnodeId));




1,642,312✔





29



  TAOS_CHECK_RETURN(tEncodeI8(pEncoder, pObj->userStopped));




1,642,312✔





30



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pObj->createTime));




1,642,312✔





31



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pObj->updateTime));




1,642,312✔





32



  TAOS_CHECK_RETURN(tEncodeCStr(pEncoder, pObj->createUser));




1,642,312✔





33



  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pObj->ownerId));




1,642,312✔





34












35



  tEndEncode(pEncoder);




821,156✔





36



  return pEncoder->pos;




821,156✔





37



}










38












39



int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj, int32_t sver) {




242,703✔





40



  int32_t code = 0;




242,703✔





41



  int32_t lino = 0;




242,703✔





42



  TAOS_CHECK_RETURN(tStartDecode(pDecoder));




242,703✔





43












44



  TAOS_CHECK_RETURN(tDecodeCStrTo(pDecoder, pObj->name));




242,703✔





45



  pObj->pCreate = taosMemoryCalloc(1, sizeof(*pObj->pCreate));




242,703✔





46



  if (NULL == pObj) {




242,703✔





47



    TAOS_CHECK_EXIT(terrno);




×







48



  }










49



  










50



  if (MND_STREAM_VER_NUMBER == sver) {




242,703✔





51



    TAOS_CHECK_RETURN(tDeserializeSCMCreateStreamReqImpl(pDecoder, pObj->pCreate));




242,703✔





52



  } else {










53



    TAOS_CHECK_RETURN(




×







54



      tDeserializeSCMCreateStreamReqImplOld(pDecoder, pObj->pCreate, 21));










55



  }










56












57



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pObj->mainSnodeId));




485,406✔





58



  TAOS_CHECK_RETURN(tDecodeI8(pDecoder, &pObj->userStopped));




485,406✔





59



  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pObj->createTime));




485,406✔





60



  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pObj->updateTime));




485,406✔





61



  if (!tDecodeIsEnd(pDecoder)) {




242,703✔





62



    TAOS_CHECK_RETURN(tDecodeCStrTo(pDecoder, pObj->createUser));




242,703✔





63



    TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pObj->ownerId));




485,406✔





64



  }










65












66



_exit:




242,703✔





67












68



  tEndDecode(pDecoder);




242,703✔





69



  tDecoderClear(pDecoder);  




242,703✔





70



  










71



  TAOS_RETURN(code);




242,703✔





72



}










73












74



void tFreeStreamObj(SStreamObj *pStream) {




452,075✔





75



  tFreeSCMCreateStreamReq(pStream->pCreate);




452,075✔





76



  taosMemoryFreeClear(pStream->pCreate);




452,075✔





77



}




452,075✔





78












79



void freeSMqConsumerEp(void* data) {




557,370✔





80



  if (data == NULL) return;




557,370✔





81



  SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)data;




557,370✔





82



  taosArrayDestroy(pConsumerEp->vgs);




557,370✔





83



  pConsumerEp->vgs = NULL;




557,370✔





84



  taosArrayDestroy(pConsumerEp->offsetRows);




557,370✔





85



  pConsumerEp->offsetRows = NULL;




557,370✔





86



}










87













UNCOV

88



static void *tDecodeSMqVgEp(const void *buf, int32_t *vgId, int8_t sver) {




×








UNCOV

89



  buf = taosDecodeFixedI32(buf, vgId);




×








UNCOV

90



  if (sver == 1) {




×








UNCOV

91



    uint64_t size = 0;




×








UNCOV

92



    buf = taosDecodeVariantU64(buf, &size);




×








UNCOV

93



    buf = POINTER_SHIFT(buf, size);




×







94



  }











UNCOV

95



  if (sver <= 3) {




×








UNCOV

96



    SEpSet epSet = {0};




×








UNCOV

97



    buf = taosDecodeSEpSet(buf, &epSet);




×







98



  }










99



  return (void *)buf;




×







100



}










101












102



static void *topicNameDup(void *p) { return taosStrdup((char *)p); }




167,383✔





103












104



int32_t tNewSMqConsumerObj(int64_t consumerId, char *cgroup, int8_t updateType, char *topic, SCMSubscribeReq *subscribe,




721,714✔





105



                           SMqConsumerObj **ppConsumer) {










106



  int32_t         code = 0;




721,714✔





107



  SMqConsumerObj *pConsumer = taosMemoryCalloc(1, sizeof(SMqConsumerObj));




721,714✔





108



  if (pConsumer == NULL) {




721,714✔






UNCOV

109



    code = terrno;




×








UNCOV

110



    goto END;




×







111



  }










112












113



  pConsumer->consumerId = consumerId;




721,714✔





114



  (void)memcpy(pConsumer->cgroup, cgroup, TSDB_CGROUP_LEN);




721,714✔





115












116



  pConsumer->epoch = 0;




721,714✔





117



  pConsumer->status = MQ_CONSUMER_STATUS_REBALANCE;




721,714✔





118



  pConsumer->hbStatus = 0;




721,714✔





119



  pConsumer->pollStatus = 0;




721,714✔





120












121



  taosInitRWLatch(&pConsumer->lock);




721,714✔





122



  pConsumer->createTime = taosGetTimestampMs();




721,714✔





123



  pConsumer->updateType = updateType;




721,714✔





124












125



  if (updateType == CONSUMER_ADD_REB) {




721,714✔





126



    pConsumer->rebNewTopics = taosArrayInit(0, sizeof(void *));




167,711✔





127



    if (pConsumer->rebNewTopics == NULL) {




167,711✔






UNCOV

128



      code = terrno;




×








UNCOV

129



      goto END;




×







130



    }










131












132



    char *topicTmp = taosStrdup(topic);




167,711✔





133



    if (taosArrayPush(pConsumer->rebNewTopics, &topicTmp) == NULL) {




335,422✔






UNCOV

134



      code = terrno;




×








UNCOV

135



      goto END;




×







136



    }










137



  } else if (updateType == CONSUMER_REMOVE_REB) {




554,003✔





138



    pConsumer->rebRemovedTopics = taosArrayInit(0, sizeof(void *));




130,801✔





139



    if (pConsumer->rebRemovedTopics == NULL) {




130,801✔






UNCOV

140



      code = terrno;




×








UNCOV

141



      goto END;




×







142



    }










143



    char *topicTmp = taosStrdup(topic);




130,801✔





144



    if (taosArrayPush(pConsumer->rebRemovedTopics, &topicTmp) == NULL) {




261,602✔





145



      code = terrno;




×








UNCOV

146



      goto END;




×







147



    }










148



  } else if (updateType == CONSUMER_INSERT_SUB) {




423,202✔





149



    tstrncpy(pConsumer->clientId, subscribe->clientId, tListLen(pConsumer->clientId));




164,607✔





150



    pConsumer->withTbName = subscribe->withTbName;




164,607✔





151



    pConsumer->autoCommit = subscribe->autoCommit;




164,607✔





152



    pConsumer->autoCommitInterval = subscribe->autoCommitInterval;




164,607✔





153



    pConsumer->resetOffsetCfg = subscribe->resetOffsetCfg;




164,607✔





154



    pConsumer->maxPollIntervalMs = subscribe->maxPollIntervalMs;




164,607✔





155



    pConsumer->sessionTimeoutMs = subscribe->sessionTimeoutMs;




164,607✔





156



    tstrncpy(pConsumer->user, subscribe->user, TSDB_USER_LEN);




164,607✔





157



    tstrncpy(pConsumer->fqdn, subscribe->fqdn, TSDB_FQDN_LEN);




164,607✔





158












159



    pConsumer->rebNewTopics = taosArrayDup(subscribe->topicNames, topicNameDup);




164,607✔





160



    if (pConsumer->rebNewTopics == NULL) {




164,607✔






UNCOV

161



      code = terrno;




×








UNCOV

162



      goto END;




×







163



    }










164



    pConsumer->assignedTopics = subscribe->topicNames;




164,607✔





165



    subscribe->topicNames = NULL;




164,607✔





166



  } else if (updateType == CONSUMER_UPDATE_SUB) {




258,595✔





167



    pConsumer->assignedTopics = subscribe->topicNames;




129,991✔





168



    subscribe->topicNames = NULL;




129,991✔





169



  }










170












171



  *ppConsumer = pConsumer;




721,714✔





172



  return 0;




721,714✔





173













UNCOV

174



END:




×








UNCOV

175



  tDeleteSMqConsumerObj(pConsumer);




×








UNCOV

176



  return code;




×







177



}










178












179



void tClearSMqConsumerObj(SMqConsumerObj *pConsumer) {




2,760,349✔





180



  if (pConsumer == NULL) return;




2,760,349✔





181



  taosArrayDestroyP(pConsumer->currentTopics, NULL);




1,450,097✔





182



  taosArrayDestroyP(pConsumer->rebNewTopics, NULL);




1,450,097✔





183



  taosArrayDestroyP(pConsumer->rebRemovedTopics, NULL);




1,450,097✔





184



  taosArrayDestroyP(pConsumer->assignedTopics, NULL);




1,450,097✔





185



}










186












187



void tDeleteSMqConsumerObj(SMqConsumerObj *pConsumer) {




2,031,966✔





188



  tClearSMqConsumerObj(pConsumer);




2,031,966✔





189



  taosMemoryFree(pConsumer);




2,031,966✔





190



}




2,031,966✔





191












192



int32_t tEncodeSMqConsumerObj(void **buf, const SMqConsumerObj *pConsumer) {




1,676,308✔





193



  int32_t tlen = 0;




1,676,308✔





194



  int32_t sz;










195



  tlen += taosEncodeFixedI64(buf, pConsumer->consumerId);




1,676,308✔





196



  tlen += taosEncodeString(buf, pConsumer->clientId);




1,676,308✔





197



  tlen += taosEncodeString(buf, pConsumer->cgroup);




1,676,308✔





198



  tlen += taosEncodeFixedI8(buf, pConsumer->updateType);




1,676,308✔





199



  tlen += taosEncodeFixedI32(buf, pConsumer->epoch);




1,676,308✔





200



  tlen += taosEncodeFixedI32(buf, pConsumer->status);




1,676,308✔





201












202



  tlen += taosEncodeFixedI32(buf, pConsumer->pid);




1,676,308✔





203



  tlen += taosEncodeFixedI64(buf, pConsumer->createTime);




1,676,308✔





204



  tlen += taosEncodeFixedI64(buf, pConsumer->subscribeTime);




1,676,308✔





205



  tlen += taosEncodeFixedI64(buf, pConsumer->rebalanceTime);




1,676,308✔





206












207



  // current topics










208



  if (pConsumer->currentTopics) {




1,676,308✔





209



    sz = taosArrayGetSize(pConsumer->currentTopics);




244,576✔





210



    tlen += taosEncodeFixedI32(buf, sz);




244,576✔





211



    for (int32_t i = 0; i < sz; i++) {




389,356✔





212



      char *topic = taosArrayGetP(pConsumer->currentTopics, i);




144,780✔





213



      tlen += taosEncodeString(buf, topic);




144,780✔





214



    }










215



  } else {










216



    tlen += taosEncodeFixedI32(buf, 0);




1,431,732✔





217



  }










218












219



  // reb new topics










220



  if (pConsumer->rebNewTopics) {




1,676,308✔





221



    sz = taosArrayGetSize(pConsumer->rebNewTopics);




1,157,498✔





222



    tlen += taosEncodeFixedI32(buf, sz);




1,157,498✔





223



    for (int32_t i = 0; i < sz; i++) {




1,906,616✔





224



      char *topic = taosArrayGetP(pConsumer->rebNewTopics, i);




749,118✔





225



      tlen += taosEncodeString(buf, topic);




749,118✔





226



    }










227



  } else {










228



    tlen += taosEncodeFixedI32(buf, 0);




518,810✔





229



  }










230












231



  // reb removed topics










232



  if (pConsumer->rebRemovedTopics) {




1,676,308✔





233



    sz = taosArrayGetSize(pConsumer->rebRemovedTopics);




754,464✔





234



    tlen += taosEncodeFixedI32(buf, sz);




754,464✔





235



    for (int32_t i = 0; i < sz; i++) {




1,269,252✔





236



      char *topic = taosArrayGetP(pConsumer->rebRemovedTopics, i);




514,788✔





237



      tlen += taosEncodeString(buf, topic);




514,788✔





238



    }










239



  } else {










240



    tlen += taosEncodeFixedI32(buf, 0);




921,844✔





241



  }










242












243



  // lost topics










244



  if (pConsumer->assignedTopics) {




1,676,308✔





245



    sz = taosArrayGetSize(pConsumer->assignedTopics);




822,076✔





246



    tlen += taosEncodeFixedI32(buf, sz);




822,076✔





247



    for (int32_t i = 0; i < sz; i++) {




1,379,860✔





248



      char *topic = taosArrayGetP(pConsumer->assignedTopics, i);




557,784✔





249



      tlen += taosEncodeString(buf, topic);




557,784✔





250



    }










251



  } else {










252



    tlen += taosEncodeFixedI32(buf, 0);




854,232✔





253



  }










254












255



  tlen += taosEncodeFixedI8(buf, pConsumer->withTbName);




1,676,308✔





256



  tlen += taosEncodeFixedI8(buf, pConsumer->autoCommit);




1,676,308✔





257



  tlen += taosEncodeFixedI32(buf, pConsumer->autoCommitInterval);




1,676,308✔





258



  tlen += taosEncodeFixedI32(buf, pConsumer->resetOffsetCfg);




1,676,308✔





259



  tlen += taosEncodeFixedI32(buf, pConsumer->maxPollIntervalMs);




1,676,308✔





260



  tlen += taosEncodeFixedI32(buf, pConsumer->sessionTimeoutMs);




1,676,308✔





261



  tlen += taosEncodeString(buf, pConsumer->user);




1,676,308✔





262



  tlen += taosEncodeString(buf, pConsumer->fqdn);




1,676,308✔





263



  return tlen;




1,676,308✔





264



}










265












266



void *tDecodeSMqConsumerObj(const void *buf, SMqConsumerObj *pConsumer, int8_t sver) {




728,383✔





267



  int32_t sz;




727,613✔





268



  buf = taosDecodeFixedI64(buf, &pConsumer->consumerId);




728,383✔





269



  buf = taosDecodeStringTo(buf, pConsumer->clientId);




728,383✔





270



  buf = taosDecodeStringTo(buf, pConsumer->cgroup);




728,383✔





271



  buf = taosDecodeFixedI8(buf, &pConsumer->updateType);




728,383✔





272



  buf = taosDecodeFixedI32(buf, &pConsumer->epoch);




728,383✔





273



  buf = taosDecodeFixedI32(buf, &pConsumer->status);




728,383✔





274












275



  buf = taosDecodeFixedI32(buf, &pConsumer->pid);




728,383✔





276



  if (sver <= 3) {




728,383✔






UNCOV

277



    SEpSet ep = {0};




×








UNCOV

278



    buf = taosDecodeSEpSet(buf, &ep);




×







279



  }










280



  buf = taosDecodeFixedI64(buf, &pConsumer->createTime);




728,383✔





281



  buf = taosDecodeFixedI64(buf, &pConsumer->subscribeTime);




728,383✔





282



  buf = taosDecodeFixedI64(buf, &pConsumer->rebalanceTime);




1,456,766✔





283












284



  // current topics










285



  buf = taosDecodeFixedI32(buf, &sz);




728,383✔





286



  pConsumer->currentTopics = taosArrayInit(sz, sizeof(void *));




728,383✔





287



  if (pConsumer->currentTopics == NULL) {




728,383✔






UNCOV

288



    return NULL;




×







289



  }










290



  for (int32_t i = 0; i < sz; i++) {




734,758✔





291



    char *topic;




6,375✔





292



    buf = taosDecodeString(buf, &topic);




6,375✔





293



    if (taosArrayPush(pConsumer->currentTopics, &topic) == NULL) {




12,750✔






UNCOV

294



      return NULL;




×







295



    }










296



  }










297












298



  // reb new topics










299



  buf = taosDecodeFixedI32(buf, &sz);




728,383✔





300



  pConsumer->rebNewTopics = taosArrayInit(sz, sizeof(void *));




728,383✔





301



  for (int32_t i = 0; i < sz; i++) {




1,067,765✔





302



    char *topic;




339,052✔





303



    buf = taosDecodeString(buf, &topic);




339,382✔





304



    if (taosArrayPush(pConsumer->rebNewTopics, &topic) == NULL) {




678,764✔






UNCOV

305



      return NULL;




×







306



    }










307



  }










308












309



  // reb removed topics










310



  buf = taosDecodeFixedI32(buf, &sz);




728,383✔





311



  pConsumer->rebRemovedTopics = taosArrayInit(sz, sizeof(void *));




728,383✔





312



  for (int32_t i = 0; i < sz; i++) {




988,156✔





313



    char *topic;




259,443✔





314



    buf = taosDecodeString(buf, &topic);




259,773✔





315



    if (taosArrayPush(pConsumer->rebRemovedTopics, &topic) == NULL) {




519,546✔






UNCOV

316



      return NULL;




×







317



    }










318



  }










319












320



  // reb removed topics










321



  buf = taosDecodeFixedI32(buf, &sz);




728,383✔





322



  pConsumer->assignedTopics = taosArrayInit(sz, sizeof(void *));




728,383✔





323



  for (int32_t i = 0; i < sz; i++) {




904,278✔





324



    char *topic;




175,730✔





325



    buf = taosDecodeString(buf, &topic);




175,895✔





326



    if (taosArrayPush(pConsumer->assignedTopics, &topic) == NULL) {




351,790✔






UNCOV

327



      return NULL;




×







328



    }










329



  }










330












331



  if (sver > 1) {




728,383✔





332



    buf = taosDecodeFixedI8(buf, &pConsumer->withTbName);




728,383✔





333



    buf = taosDecodeFixedI8(buf, &pConsumer->autoCommit);




728,383✔





334



    buf = taosDecodeFixedI32(buf, &pConsumer->autoCommitInterval);




728,383✔





335



    buf = taosDecodeFixedI32(buf, &pConsumer->resetOffsetCfg);




1,456,766✔





336



  }










337



  if (sver > 2) {




728,383✔





338



    buf = taosDecodeFixedI32(buf, &pConsumer->maxPollIntervalMs);




728,383✔





339



    buf = taosDecodeFixedI32(buf, &pConsumer->sessionTimeoutMs);




728,383✔





340



    buf = taosDecodeStringTo(buf, pConsumer->user);




728,383✔





341



    buf = taosDecodeStringTo(buf, pConsumer->fqdn);




1,456,766✔





342



  } else {











UNCOV

343



    pConsumer->maxPollIntervalMs = DEFAULT_MAX_POLL_INTERVAL;




×








UNCOV

344



    pConsumer->sessionTimeoutMs = DEFAULT_SESSION_TIMEOUT;




×







345



  }










346












347



  return (void *)buf;




728,383✔





348



}










349












350



int32_t tEncodeOffRows(void **buf, SArray *offsetRows) {




1,484,396✔





351



  int32_t tlen = 0;




1,484,396✔





352



  int32_t szVgs = taosArrayGetSize(offsetRows);




1,484,396✔





353



  tlen += taosEncodeFixedI32(buf, szVgs);




1,484,396✔





354



  for (int32_t j = 0; j < szVgs; ++j) {




3,296,332✔





355



    OffsetRows *offRows = taosArrayGet(offsetRows, j);




1,811,936✔





356



    tlen += taosEncodeFixedI32(buf, offRows->vgId);




1,811,936✔





357



    tlen += taosEncodeFixedI64(buf, offRows->rows);




1,811,936✔





358



    tlen += taosEncodeFixedI8(buf, offRows->offset.type);




1,811,936✔





359



    if (offRows->offset.type == TMQ_OFFSET__SNAPSHOT_DATA || offRows->offset.type == TMQ_OFFSET__SNAPSHOT_META) {




1,811,936✔





360



      tlen += taosEncodeFixedI64(buf, offRows->offset.uid);




61,256✔





361



      tlen += taosEncodeFixedI64(buf, offRows->offset.ts);




122,512✔





362



    } else if (offRows->offset.type == TMQ_OFFSET__LOG) {




1,750,680✔





363



      tlen += taosEncodeFixedI64(buf, offRows->offset.version);




3,157,324✔





364



    } else {










365



      // do nothing










366



    }










367



    tlen += taosEncodeFixedI64(buf, offRows->ever);




3,623,872✔





368



  }










369












370



  return tlen;




1,484,396✔





371



}










372












373



int32_t tEncodeSMqConsumerEp(void **buf, const SMqConsumerEp *pConsumerEp) {




502,914✔





374



  int32_t tlen = 0;




502,914✔





375



  tlen += taosEncodeFixedI64(buf, pConsumerEp->consumerId);




502,914✔





376



  int32_t sz = taosArrayGetSize(pConsumerEp->vgs);




502,914✔





377



  tlen += taosEncodeFixedI32(buf, sz);




502,914✔





378



  for (int32_t i = 0; i < sz; i++) {




1,607,010✔





379



    int32_t* vgId = (int32_t*)taosArrayGet(pConsumerEp->vgs, i);




1,104,096✔





380



    if (vgId == NULL) continue;




1,104,096✔





381



    tlen += taosEncodeFixedI32(buf, *vgId);




2,208,192✔





382



  }










383












384



  return tlen + tEncodeOffRows(buf, pConsumerEp->offsetRows);




502,914✔





385



}










386












387



void *tDecodeOffRows(const void *buf, SArray **offsetRows, int8_t sver) {




581,414✔





388



  int32_t szVgs = 0;




581,414✔





389



  buf = taosDecodeFixedI32(buf, &szVgs);




581,414✔





390



  if (szVgs > 0) {




581,414✔





391



    *offsetRows = taosArrayInit(szVgs, sizeof(OffsetRows));




249,329✔





392



    if (NULL == *offsetRows) return NULL;




249,329✔





393



    for (int32_t j = 0; j < szVgs; ++j) {




1,047,333✔





394



      OffsetRows *offRows = taosArrayReserve(*offsetRows, 1);




798,004✔





395



      buf = taosDecodeFixedI32(buf, &offRows->vgId);




798,004✔





396



      buf = taosDecodeFixedI64(buf, &offRows->rows);




798,004✔





397



      buf = taosDecodeFixedI8(buf, &offRows->offset.type);




798,004✔





398



      if (offRows->offset.type == TMQ_OFFSET__SNAPSHOT_DATA || offRows->offset.type == TMQ_OFFSET__SNAPSHOT_META) {




798,004✔





399



        buf = taosDecodeFixedI64(buf, &offRows->offset.uid);




30,572✔





400



        buf = taosDecodeFixedI64(buf, &offRows->offset.ts);




61,144✔





401



      } else if (offRows->offset.type == TMQ_OFFSET__LOG) {




767,432✔





402



        buf = taosDecodeFixedI64(buf, &offRows->offset.version);




1,374,152✔





403



      } else {










404



        // do nothing










405



      }










406



      if (sver > 2) {




798,004✔





407



        buf = taosDecodeFixedI64(buf, &offRows->ever);




1,596,008✔





408



      }










409



    }










410



  }










411



  return (void *)buf;




581,414✔





412



}










413












414



void *tDecodeSMqConsumerEp(const void *buf, SMqConsumerEp *pConsumerEp, int8_t sver) {




187,250✔





415



  buf = taosDecodeFixedI64(buf, &pConsumerEp->consumerId);




187,250✔





416



  int32_t sz = 0;




187,250✔





417



  buf = taosDecodeFixedI32(buf, &sz);




187,250✔





418



  pConsumerEp->vgs = taosArrayInit(sz, sizeof(int32_t));




187,250✔





419



  if (pConsumerEp->vgs == NULL) {




187,250✔






UNCOV

420



    return NULL;




×







421



  }










422



  for (int32_t i = 0; i < sz; i++) {




673,583✔





423



    int32_t* vgId = taosArrayReserve(pConsumerEp->vgs, 1);




486,333✔





424



    if (vgId == NULL) {




486,333✔






UNCOV

425



      return NULL;




×







426



    }










427



    if (sver <= 3) {




486,333✔






UNCOV

428



      buf = tDecodeSMqVgEp(buf, vgId, sver);




×







429



    } else {










430



      buf = taosDecodeFixedI32(buf, vgId);




486,333✔





431



    }










432



  }










433



  if (sver > 1) {




187,250✔





434



    buf = tDecodeOffRows(buf, &pConsumerEp->offsetRows, sver);




187,250✔





435



  }










436












437



  return (void *)buf;




187,250✔





438



}










439












440



int32_t tNewSubscribeObj(const char *key, SMqSubscribeObj **ppSub) {




137,535✔





441



  int32_t          code = 0;




137,535✔





442



  int32_t          lino = 0;




137,535✔





443



  PRINT_LOG_START




137,535✔





444



  SMqSubscribeObj *pSubObj = taosMemoryCalloc(1, sizeof(SMqSubscribeObj));




137,535✔





445



  MND_TMQ_NULL_CHECK(pSubObj);




137,535✔





446



  










447



  (void)memcpy(pSubObj->key, key, TSDB_SUBSCRIBE_KEY_LEN);




137,535✔





448



  taosInitRWLatch(&pSubObj->lock);




137,535✔





449



  pSubObj->vgNum = 0;




137,535✔





450



  pSubObj->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);




137,535✔





451



  MND_TMQ_NULL_CHECK(pSubObj->consumerHash);




137,535✔





452



  taosHashSetFreeFp(pSubObj->consumerHash, freeSMqConsumerEp);




137,535✔





453












454



  pSubObj->unassignedVgs = taosArrayInit(0, sizeof(int32_t));




137,535✔





455



  MND_TMQ_NULL_CHECK(pSubObj->unassignedVgs);




137,535✔





456



  *ppSub = pSubObj;




137,535✔





457



  pSubObj = NULL;




137,535✔





458












459



END:




137,535✔





460



  tDeleteSubscribeObj(pSubObj);




137,535✔





461



  PRINT_LOG_END




137,535✔





462



  return code;




137,535✔





463



}










464












465



int32_t tCloneSubscribeObj(SMqSubscribeObj *pSub, SMqSubscribeObj **ppSub) {




220,215✔





466



  int32_t          code = 0;




220,215✔





467



  int32_t          lino = 0;




220,215✔





468



  PRINT_LOG_START




220,215✔





469



  taosRLockLatch(&pSub->lock);




220,215✔





470












471



  SMqSubscribeObj *pSubNew = taosMemoryCalloc(1, sizeof(SMqSubscribeObj));




220,215✔





472



  MND_TMQ_NULL_CHECK(pSubNew);




220,215✔





473



  *ppSub = pSubNew;




220,215✔





474












475



  (void)memcpy(pSubNew->key, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);




220,215✔





476



  taosInitRWLatch(&pSubNew->lock);




220,215✔





477












478



  pSubNew->dbUid = pSub->dbUid;




220,215✔





479



  pSubNew->stbUid = pSub->stbUid;




220,215✔





480



  pSubNew->subType = pSub->subType;




220,215✔





481



  pSubNew->withMeta = pSub->withMeta;




220,215✔





482












483



  pSubNew->vgNum = pSub->vgNum;




220,215✔





484



  pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);




220,215✔





485



  MND_TMQ_NULL_CHECK(pSubNew->consumerHash);




220,215✔





486



  taosHashSetFreeFp(pSubNew->consumerHash, freeSMqConsumerEp);




220,215✔





487












488



  void          *pIter = NULL;




220,215✔





489



  SMqConsumerEp *pConsumerEp = NULL;




220,215✔





490



  while (1) {




202,058✔





491



    pIter = taosHashIterate(pSub->consumerHash, pIter);




422,273✔





492



    if (pIter == NULL) break;




422,273✔





493



    pConsumerEp = (SMqConsumerEp *)pIter;




202,058✔





494



    SMqConsumerEp newEp = {




202,223✔





495



        .consumerId = pConsumerEp->consumerId,




202,058✔





496



        .vgs = taosArrayDup(pConsumerEp->vgs, NULL),




202,058✔





497



    };










498



    MND_TMQ_RETURN_CHECK(taosHashPut(pSubNew->consumerHash, &newEp.consumerId, sizeof(int64_t), &newEp, sizeof(SMqConsumerEp)));




202,058✔





499



  }










500



  pSubNew->unassignedVgs = taosArrayDup(pSub->unassignedVgs, NULL);




220,215✔





501



  MND_TMQ_NULL_CHECK(pSubNew->unassignedVgs);




220,215✔





502



  










503



  pSubNew->offsetRows = taosArrayDup(pSub->offsetRows, NULL);




220,215✔





504



  MND_TMQ_CONDITION_CHECK(pSub->offsetRows == NULL || pSubNew->offsetRows != NULL, terrno);




220,215✔





505



  (void)memcpy(pSubNew->dbName, pSub->dbName, TSDB_DB_FNAME_LEN);




220,215✔





506












507



END:




220,215✔





508



  taosRUnLockLatch(&pSub->lock);




220,215✔





509



  return code;




220,215✔





510



}










511












512



void tDeleteSubscribeObj(SMqSubscribeObj *pSub) {




898,839✔





513



  if (pSub == NULL) return;




898,839✔





514



  taosHashCleanup(pSub->consumerHash);




751,914✔





515



  pSub->consumerHash = NULL;




751,914✔





516



  taosArrayDestroy(pSub->unassignedVgs);




751,914✔





517



  pSub->unassignedVgs = NULL;




751,914✔





518



  taosArrayDestroy(pSub->offsetRows);




751,914✔





519



  pSub->offsetRows = NULL;




751,914✔





520



}










521












522



int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {




981,482✔





523



  int32_t tlen = 0;




981,482✔





524



  tlen += taosEncodeString(buf, pSub->key);




981,482✔





525



  tlen += taosEncodeFixedI64(buf, pSub->dbUid);




981,482✔





526



  tlen += taosEncodeFixedI32(buf, pSub->vgNum);




981,482✔





527



  tlen += taosEncodeFixedI8(buf, pSub->subType);




981,482✔





528



  tlen += taosEncodeFixedI8(buf, pSub->withMeta);




981,482✔





529



  tlen += taosEncodeFixedI64(buf, pSub->stbUid);




981,482✔





530












531



  void   *pIter = NULL;




981,482✔





532



  int32_t sz = taosHashGetSize(pSub->consumerHash);




981,482✔





533



  tlen += taosEncodeFixedI32(buf, sz);




981,482✔





534












535



  int32_t cnt = 0;




981,482✔





536



  while (1) {




502,914✔





537



    pIter = taosHashIterate(pSub->consumerHash, pIter);




1,484,396✔





538



    if (pIter == NULL) break;




1,484,396✔





539



    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;




502,914✔





540



    tlen += tEncodeSMqConsumerEp(buf, pConsumerEp);




502,914✔





541



    cnt++;




502,914✔





542



  }










543



  if (cnt != sz) return -1;




981,482✔





544



  int32_t len = taosArrayGetSize(pSub->unassignedVgs);




981,482✔





545



  tlen += taosEncodeFixedI32(buf, len);




981,482✔





546



  for (int32_t i = 0; i < len; i++) {




2,430,640✔





547



    int32_t* vgId = (int32_t*)taosArrayGet(pSub->unassignedVgs, i);




1,449,158✔





548



    if (vgId == NULL) continue;




1,449,158✔





549



    tlen += taosEncodeFixedI32(buf, *vgId);




2,898,316✔





550



  }










551



  tlen += taosEncodeString(buf, pSub->dbName);




981,482✔





552












553



  tlen += tEncodeOffRows(buf, pSub->offsetRows);




981,482✔





554



  return tlen;




981,482✔





555



}










556












557



void *tDecodeSubscribeObj(const void *buf, SMqSubscribeObj *pSub, int8_t sver) {




394,164✔





558



  //










559



  buf = taosDecodeStringTo(buf, pSub->key);




394,164✔





560



  buf = taosDecodeFixedI64(buf, &pSub->dbUid);




394,164✔





561



  buf = taosDecodeFixedI32(buf, &pSub->vgNum);




394,164✔





562



  buf = taosDecodeFixedI8(buf, &pSub->subType);




394,164✔





563



  buf = taosDecodeFixedI8(buf, &pSub->withMeta);




394,164✔





564



  buf = taosDecodeFixedI64(buf, &pSub->stbUid);




394,494✔





565












566



  int32_t sz;




393,834✔





567



  buf = taosDecodeFixedI32(buf, &sz);




394,164✔





568












569



  pSub->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);




394,164✔





570



  if (pSub->consumerHash == NULL) {




394,164✔






UNCOV

571



    return NULL;




×







572



  }










573



  taosHashSetFreeFp(pSub->consumerHash, freeSMqConsumerEp);




394,164✔





574












575



  for (int32_t i = 0; i < sz; i++) {




581,414✔





576



    SMqConsumerEp consumerEp = {0};




187,250✔





577



    buf = tDecodeSMqConsumerEp(buf, &consumerEp, sver);




187,250✔





578



    if (taosHashPut(pSub->consumerHash, &consumerEp.consumerId, sizeof(int64_t), &consumerEp, sizeof(SMqConsumerEp)) !=




187,250✔





579



        0)











UNCOV

580



      return NULL;




×







581



  }










582












583



  int32_t len = 0;




394,164✔





584



  buf = taosDecodeFixedI32(buf, &len);




394,164✔





585



  pSub->unassignedVgs = taosArrayInit(len, sizeof(int32_t));




394,164✔





586



  if (pSub->unassignedVgs == NULL) {




394,164✔






UNCOV

587



    return NULL;




×







588



  }










589



  for (int32_t i = 0; i < len; i++) {




1,066,380✔





590



    int32_t* vgId = taosArrayReserve(pSub->unassignedVgs, 1);




672,216✔





591



    if (vgId == NULL) {




672,216✔






UNCOV

592



      return NULL;




×







593



    }










594



    if (sver <= 3) {




672,216✔






UNCOV

595



      buf = tDecodeSMqVgEp(buf, vgId, sver);




×







596



    } else {










597



      buf = taosDecodeFixedI32(buf, vgId);




672,216✔





598



    }










599



  }










600












601



  buf = taosDecodeStringTo(buf, pSub->dbName);




394,164✔





602












603



  if (sver > 1) {




394,164✔





604



    buf = tDecodeOffRows(buf, &pSub->offsetRows, sver);




394,164✔





605



  }










606



  return (void *)buf;




394,164✔





607



}










608












609



int32_t mndInitConfigObj(SConfigItem *pItem, SConfigObj *pObj) {




39,501,891✔





610



  tstrncpy(pObj->name, pItem->name, CFG_NAME_MAX_LEN);




39,501,891✔





611



  pObj->dtype = pItem->dtype;




39,501,891✔





612



  switch (pItem->dtype) {




39,501,891✔






UNCOV

613



    case CFG_DTYPE_NONE:




×








UNCOV

614



      break;




×







615



    case CFG_DTYPE_BOOL:




9,791,067✔





616



      pObj->bval = pItem->bval;




9,791,067✔





617



      break;




9,791,067✔





618



    case CFG_DTYPE_INT32:




20,257,380✔





619



      pObj->i32 = pItem->i32;




20,257,380✔





620



      break;




20,257,380✔





621



    case CFG_DTYPE_INT64:




2,363,361✔





622



      pObj->i64 = pItem->i64;




2,363,361✔





623



      break;




2,363,361✔





624



    case CFG_DTYPE_FLOAT:




675,246✔





625



    case CFG_DTYPE_DOUBLE:










626



      pObj->fval = pItem->fval;




675,246✔





627



      break;




675,246✔





628



    case CFG_DTYPE_STRING:




6,414,837✔





629



    case CFG_DTYPE_DIR:










630



    case CFG_DTYPE_LOCALE:










631



    case CFG_DTYPE_CHARSET:










632



    case CFG_DTYPE_TIMEZONE:










633



      pObj->str = taosStrdup(pItem->str);




6,414,837✔





634



      if (pObj->str == NULL) {




6,414,837✔






UNCOV

635



        taosMemoryFree(pObj);




×








UNCOV

636



        return TSDB_CODE_OUT_OF_MEMORY;




×







637



      }










638



      break;




6,414,837✔





639



  }










640



  return TSDB_CODE_SUCCESS;




39,501,891✔





641



}










642












643



int32_t mndUpdateObj(SConfigObj *pObjNew, const char *name, char *value) {




14,173✔





644



  int32_t code = 0;




14,173✔





645



  switch (pObjNew->dtype) {




14,173✔





646



    case CFG_DTYPE_BOOL: {




4,007✔





647



      bool tmp = false;




4,007✔





648



      if (strcasecmp(value, "true") == 0) {




4,007✔






UNCOV

649



        tmp = true;




×







650



      }










651



      if (taosStr2Int32(value, NULL, 10) > 0) {




4,007✔





652



        tmp = true;




2,338✔





653



      }










654



      pObjNew->bval = tmp;




4,007✔





655



      break;




4,007✔





656



    }










657



    case CFG_DTYPE_INT32: {




9,572✔





658



      int32_t ival;




9,572✔





659



      TAOS_CHECK_RETURN(taosStrHumanToInt32(value, &ival));




9,572✔





660



      pObjNew->i32 = ival;




9,572✔





661



      break;




9,572✔





662



    }










663



    case CFG_DTYPE_INT64: {




524✔





664



      int64_t ival;




524✔





665



      TAOS_CHECK_RETURN(taosStrHumanToInt64(value, &ival));




524✔





666



      pObjNew->i64 = ival;




524✔





667



      break;




524✔





668



    }











UNCOV

669



    case CFG_DTYPE_FLOAT:




×







670



    case CFG_DTYPE_DOUBLE: {










671



      float dval = 0;




×







672



      TAOS_CHECK_RETURN(parseCfgReal(value, &dval));




×








UNCOV

673



      pObjNew->fval = dval;




×








UNCOV

674



      break;




×







675



    }










676



    case CFG_DTYPE_DIR:




70✔





677



    case CFG_DTYPE_TIMEZONE:










678



    case CFG_DTYPE_CHARSET:










679



    case CFG_DTYPE_LOCALE:










680



    case CFG_DTYPE_STRING: {










681



      pObjNew->str = taosStrdup(value);




70✔





682



      if (pObjNew->str == NULL) {




70✔






UNCOV

683



        code = terrno;




×








UNCOV

684



        return code;




×







685



      }










686



      break;




70✔





687



    }











UNCOV

688



    case CFG_DTYPE_NONE:




×








UNCOV

689



      break;




×








UNCOV

690



    default:




×








UNCOV

691



      code = TSDB_CODE_INVALID_CFG;




×








UNCOV

692



      break;




×







693



  }










694



  return code;




14,173✔





695



}










696












697



SConfigObj mndInitConfigVersion() {




338,415✔





698



  SConfigObj obj;




337,492✔





699



  memset(&obj, 0, sizeof(SConfigObj));




338,415✔





700












701



  tstrncpy(obj.name, "tsmmConfigVersion", CFG_NAME_MAX_LEN);




338,415✔





702



  obj.dtype = CFG_DTYPE_INT32;




338,415✔





703



  obj.i32 = 0;




338,415✔





704



  return obj;




338,415✔





705



}










706












707



int32_t tEncodeSConfigObj(SEncoder *pEncoder, const SConfigObj *pObj) {




292,271,700✔





708



  TAOS_CHECK_RETURN(tStartEncode(pEncoder));




292,271,700✔





709



  TAOS_CHECK_RETURN(tEncodeCStr(pEncoder, pObj->name));




584,543,400✔





710












711



  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pObj->dtype));




584,543,400✔





712



  switch (pObj->dtype) {




292,271,700✔





713



    case CFG_DTYPE_BOOL:




71,817,320✔





714



      TAOS_CHECK_RETURN(tEncodeI8(pEncoder, pObj->bval));




143,634,640✔





715



      break;




71,817,320✔





716



    case CFG_DTYPE_INT32:




151,127,756✔





717



      TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pObj->i32));




302,255,512✔





718



      break;




151,127,756✔





719



    case CFG_DTYPE_INT64:




17,333,158✔





720



      TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pObj->i64));




34,666,316✔





721



      break;




17,333,158✔





722



    case CFG_DTYPE_FLOAT:




4,951,732✔





723



    case CFG_DTYPE_DOUBLE:










724



      TAOS_CHECK_RETURN(tEncodeFloat(pEncoder, pObj->fval));




9,903,464✔





725



      break;




4,951,732✔





726



    case CFG_DTYPE_STRING:




47,041,734✔





727



    case CFG_DTYPE_DIR:










728



    case CFG_DTYPE_LOCALE:










729



    case CFG_DTYPE_CHARSET:










730



    case CFG_DTYPE_TIMEZONE:










731



      if (pObj->str != NULL) {




47,041,734✔





732



        TAOS_CHECK_RETURN(tEncodeCStr(pEncoder, pObj->str));




94,083,468✔





733



      } else {











UNCOV

734



        TAOS_CHECK_RETURN(tEncodeCStr(pEncoder, ""));




×







735



      }










736



      break;




47,041,734✔





737



    default:




×







738



      break;




×







739



  }










740



  tEndEncode(pEncoder);




292,271,700✔





741



  return pEncoder->pos;




292,271,700✔





742



}










743












744



int32_t tDecodeSConfigObj(SDecoder *pDecoder, SConfigObj *pObj) {




55,337,154✔





745



  TAOS_CHECK_RETURN(tStartDecode(pDecoder));




55,337,154✔





746



  TAOS_CHECK_RETURN(tDecodeCStrTo(pDecoder, pObj->name));




55,337,154✔





747



  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, (int32_t *)&pObj->dtype));




110,674,308✔





748



  switch (pObj->dtype) {




55,337,154✔






UNCOV

749



    case CFG_DTYPE_NONE:




×








UNCOV

750



      break;




×







751



    case CFG_DTYPE_BOOL:




13,584,287✔





752



      TAOS_CHECK_RETURN(tDecodeBool(pDecoder, &pObj->bval));




13,584,287✔





753



      break;




13,584,287✔





754



    case CFG_DTYPE_INT32:




28,644,967✔





755



      TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pObj->i32));




57,289,934✔





756



      break;




28,644,967✔





757



    case CFG_DTYPE_INT64:




3,279,111✔





758



      TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pObj->i64));




6,558,222✔





759



      break;




3,279,111✔





760



    case CFG_DTYPE_FLOAT:




936,906✔





761



    case CFG_DTYPE_DOUBLE:










762



      TAOS_CHECK_RETURN(tDecodeFloat(pDecoder, &pObj->fval));




1,873,812✔





763



      break;




936,906✔





764



    case CFG_DTYPE_STRING:




8,891,883✔





765



    case CFG_DTYPE_DIR:










766



    case CFG_DTYPE_LOCALE:










767



    case CFG_DTYPE_CHARSET:










768



    case CFG_DTYPE_TIMEZONE:










769



      TAOS_CHECK_RETURN(tDecodeCStrAlloc(pDecoder, &pObj->str));




17,783,766✔





770



      break;




8,891,883✔





771



  }










772



  tEndDecode(pDecoder);




55,337,154✔





773



  TAOS_RETURN(TSDB_CODE_SUCCESS);




55,337,154✔





774



}










775












776



void tFreeSConfigObj(SConfigObj *obj) {




95,200,260✔





777



  if (obj == NULL) {




95,200,260✔






UNCOV

778



    return;




×







779



  }










780



  if (obj->dtype == CFG_DTYPE_STRING || obj->dtype == CFG_DTYPE_DIR || obj->dtype == CFG_DTYPE_LOCALE ||




95,200,260✔





781



      obj->dtype == CFG_DTYPE_CHARSET || obj->dtype == CFG_DTYPE_TIMEZONE) {




81,505,523✔





782



    taosMemoryFree(obj->str);




15,305,897✔





783



  }










784



}










785












786



// SMqSubActionLogEntry *tCloneSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {










787



//   SMqSubActionLogEntry *pEntryNew = taosMemoryMalloc(sizeof(SMqSubActionLogEntry));










788



//   if (pEntryNew == NULL) return NULL;










789



//   pEntryNew->epoch = pEntry->epoch;










790



//   pEntryNew->consumers = taosArrayDup(pEntry->consumers, (__array_item_dup_fn_t)tCloneSMqConsumerEp);










791



//   return pEntryNew;










792



// }










793



//










794



// void tDeleteSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {










795



//   taosArrayDestroyEx(pEntry->consumers, (FDelete)tDeleteSMqConsumerEp);










796



// }










797












798



// int32_t tEncodeSMqSubActionLogEntry(void **buf, const SMqSubActionLogEntry *pEntry) {










799



//   int32_t tlen = 0;










800



//   tlen += taosEncodeFixedI32(buf, pEntry->epoch);










801



//   tlen += taosEncodeArray(buf, pEntry->consumers, (FEncode)tEncodeSMqSubActionLogEntry);










802



//   return tlen;










803



// }










804



//










805



// void *tDecodeSMqSubActionLogEntry(const void *buf, SMqSubActionLogEntry *pEntry) {










806



//   buf = taosDecodeFixedI32(buf, &pEntry->epoch);










807



//   buf = taosDecodeArray(buf, &pEntry->consumers, (FDecode)tDecodeSMqSubActionLogEntry, sizeof(SMqSubActionLogEntry));










808



//   return (void *)buf;










809



// }










810












811



// SMqSubActionLogObj *tCloneSMqSubActionLogObj(SMqSubActionLogObj *pLog) {










812



//   SMqSubActionLogObj *pLogNew = taosMemoryMalloc(sizeof(SMqSubActionLogObj));










813



//   if (pLogNew == NULL) return pLogNew;










814



//   memcpy(pLogNew->key, pLog->key, TSDB_SUBSCRIBE_KEY_LEN);










815



//   pLogNew->logs = taosArrayDup(pLog->logs, (__array_item_dup_fn_t)tCloneSMqConsumerEp);










816



//   return pLogNew;










817



// }










818



//










819



// void tDeleteSMqSubActionLogObj(SMqSubActionLogObj *pLog) {










820



//   taosArrayDestroyEx(pLog->logs, (FDelete)tDeleteSMqConsumerEp);










821



// }










822












823



// int32_t tEncodeSMqSubActionLogObj(void **buf, const SMqSubActionLogObj *pLog) {










824



//   int32_t tlen = 0;










825



//   tlen += taosEncodeString(buf, pLog->key);










826



//   tlen += taosEncodeArray(buf, pLog->logs, (FEncode)tEncodeSMqSubActionLogEntry);










827



//   return tlen;










828



// }










829



//










830



// void *tDecodeSMqSubActionLogObj(const void *buf, SMqSubActionLogObj *pLog) {










831



//   buf = taosDecodeStringTo(buf, pLog->key);










832



//   buf = taosDecodeArray(buf, &pLog->logs, (FDecode)tDecodeSMqSubActionLogEntry, sizeof(SMqSubActionLogEntry));










833



//   return (void *)buf;










834



// }










835



//










836



// int32_t tEncodeSMqOffsetObj(void **buf, const SMqOffsetObj *pOffset) {










837



//   int32_t tlen = 0;










838



//   tlen += taosEncodeString(buf, pOffset->key);










839



//   tlen += taosEncodeFixedI64(buf, pOffset->offset);










840



//   return tlen;










841



// }










842



//










843



// void *tDecodeSMqOffsetObj(void *buf, SMqOffsetObj *pOffset) {










844



//   buf = taosDecodeStringTo(buf, pOffset->key);










845



//   buf = taosDecodeFixedI64(buf, &pOffset->offset);










846



//   return buf;










847



// }
























    

  • 

    •