• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #4863

26 Nov 2025 05:46AM UTC coverage: 64.539% (+0.2%) from 64.294%
#4863

push

travis-ci

guanshengliang
Merge branch '3.0' into cover/3.0

767 of 945 new or added lines in 33 files covered. (81.16%)

701 existing lines in 109 files now uncovered.

158204 of 245129 relevant lines covered (64.54%)

111903199.95 hits per line

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

87.8
/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) {
795,728✔
23
  TAOS_CHECK_RETURN(tStartEncode(pEncoder));
795,728✔
24

25
  TAOS_CHECK_RETURN(tEncodeCStr(pEncoder, pObj->name));
1,591,456✔
26
  TAOS_CHECK_RETURN(tSerializeSCMCreateStreamReqImpl(pEncoder, pObj->pCreate));
795,728✔
27

28
  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pObj->mainSnodeId));
1,591,456✔
29
  TAOS_CHECK_RETURN(tEncodeI8(pEncoder, pObj->userStopped));
1,591,456✔
30
  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pObj->createTime));
1,591,456✔
31
  TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pObj->updateTime));
1,591,456✔
32

33
  tEndEncode(pEncoder);
795,728✔
34
  return pEncoder->pos;
795,728✔
35
}
36

37
int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj, int32_t sver) {
217,505✔
38
  int32_t code = 0;
217,505✔
39
  int32_t lino = 0;
217,505✔
40
  TAOS_CHECK_RETURN(tStartDecode(pDecoder));
217,505✔
41

42
  TAOS_CHECK_RETURN(tDecodeCStrTo(pDecoder, pObj->name));
217,505✔
43
  pObj->pCreate = taosMemoryCalloc(1, sizeof(*pObj->pCreate));
217,505✔
44
  if (NULL == pObj) {
217,505✔
45
    TAOS_CHECK_EXIT(terrno);
×
46
  }
47
  
48
  if (MND_STREAM_VER_NUMBER == sver) {
217,505✔
49
    TAOS_CHECK_RETURN(tDeserializeSCMCreateStreamReqImpl(pDecoder, pObj->pCreate));
217,505✔
50
  } else {
NEW
51
    TAOS_CHECK_RETURN(
×
52
      tDeserializeSCMCreateStreamReqImplOld(pDecoder, pObj->pCreate, 21));
53
  }
54

55
  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pObj->mainSnodeId));
435,010✔
56
  TAOS_CHECK_RETURN(tDecodeI8(pDecoder, &pObj->userStopped));
435,010✔
57
  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pObj->createTime));
435,010✔
58
  TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pObj->updateTime));
435,010✔
59

60
_exit:
217,505✔
61

62
  tEndDecode(pDecoder);
217,505✔
63
  tDecoderClear(pDecoder);  
217,505✔
64
  
65
  TAOS_RETURN(code);
217,505✔
66
}
67

68
void tFreeStreamObj(SStreamObj *pStream) {
438,638✔
69
  tFreeSCMCreateStreamReq(pStream->pCreate);
438,638✔
70
  taosMemoryFreeClear(pStream->pCreate);
438,638✔
71
}
438,638✔
72

73
void freeSMqConsumerEp(void* data) {
161,854✔
74
  if (data == NULL) return;
161,854✔
75
  SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)data;
161,854✔
76
  taosArrayDestroy(pConsumerEp->vgs);
161,854✔
77
  pConsumerEp->vgs = NULL;
161,854✔
78
  taosArrayDestroy(pConsumerEp->offsetRows);
161,854✔
79
  pConsumerEp->offsetRows = NULL;
161,854✔
80
}
81

82
int32_t tEncodeSMqVgEp(void **buf, const SMqVgEp *pVgEp) {
947,662✔
83
  int32_t tlen = 0;
947,662✔
84
  tlen += taosEncodeFixedI32(buf, pVgEp->vgId);
947,662✔
85
  tlen += taosEncodeSEpSet(buf, &pVgEp->epSet);
947,662✔
86
  return tlen;
947,662✔
87
}
88

89
void *tDecodeSMqVgEp(const void *buf, SMqVgEp *pVgEp, int8_t sver) {
434,691✔
90
  buf = taosDecodeFixedI32(buf, &pVgEp->vgId);
434,691✔
91
  if (sver == 1) {
434,691✔
92
    uint64_t size = 0;
×
93
    buf = taosDecodeVariantU64(buf, &size);
×
94
    buf = POINTER_SHIFT(buf, size);
×
95
  }
96
  buf = taosDecodeSEpSet(buf, &pVgEp->epSet);
434,691✔
97
  return (void *)buf;
434,691✔
98
}
99

100
static void *topicNameDup(void *p) { return taosStrdup((char *)p); }
52,141✔
101

102
int32_t tNewSMqConsumerObj(int64_t consumerId, char *cgroup, int8_t updateType, char *topic, SCMSubscribeReq *subscribe,
227,605✔
103
                           SMqConsumerObj **ppConsumer) {
104
  int32_t         code = 0;
227,605✔
105
  SMqConsumerObj *pConsumer = taosMemoryCalloc(1, sizeof(SMqConsumerObj));
227,605✔
106
  if (pConsumer == NULL) {
227,605✔
107
    code = terrno;
×
108
    goto END;
×
109
  }
110

111
  pConsumer->consumerId = consumerId;
227,605✔
112
  (void)memcpy(pConsumer->cgroup, cgroup, TSDB_CGROUP_LEN);
227,605✔
113

114
  pConsumer->epoch = 0;
227,605✔
115
  pConsumer->status = MQ_CONSUMER_STATUS_REBALANCE;
227,605✔
116
  pConsumer->hbStatus = 0;
227,605✔
117
  pConsumer->pollStatus = 0;
227,605✔
118

119
  taosInitRWLatch(&pConsumer->lock);
227,605✔
120
  pConsumer->createTime = taosGetTimestampMs();
227,605✔
121
  pConsumer->updateType = updateType;
227,605✔
122

123
  if (updateType == CONSUMER_ADD_REB) {
227,605✔
124
    pConsumer->rebNewTopics = taosArrayInit(0, sizeof(void *));
52,349✔
125
    if (pConsumer->rebNewTopics == NULL) {
52,349✔
126
      code = terrno;
×
127
      goto END;
×
128
    }
129

130
    char *topicTmp = taosStrdup(topic);
52,349✔
131
    if (taosArrayPush(pConsumer->rebNewTopics, &topicTmp) == NULL) {
104,698✔
132
      code = terrno;
×
133
      goto END;
×
134
    }
135
  } else if (updateType == CONSUMER_REMOVE_REB) {
175,256✔
136
    pConsumer->rebRemovedTopics = taosArrayInit(0, sizeof(void *));
41,119✔
137
    if (pConsumer->rebRemovedTopics == NULL) {
41,119✔
138
      code = terrno;
×
139
      goto END;
×
140
    }
141
    char *topicTmp = taosStrdup(topic);
41,119✔
142
    if (taosArrayPush(pConsumer->rebRemovedTopics, &topicTmp) == NULL) {
82,238✔
143
      code = terrno;
×
144
      goto END;
×
145
    }
146
  } else if (updateType == CONSUMER_INSERT_SUB) {
134,137✔
147
    tstrncpy(pConsumer->clientId, subscribe->clientId, tListLen(pConsumer->clientId));
51,630✔
148
    pConsumer->withTbName = subscribe->withTbName;
51,630✔
149
    pConsumer->autoCommit = subscribe->autoCommit;
51,630✔
150
    pConsumer->autoCommitInterval = subscribe->autoCommitInterval;
51,630✔
151
    pConsumer->resetOffsetCfg = subscribe->resetOffsetCfg;
51,630✔
152
    pConsumer->maxPollIntervalMs = subscribe->maxPollIntervalMs;
51,630✔
153
    pConsumer->sessionTimeoutMs = subscribe->sessionTimeoutMs;
51,630✔
154
    tstrncpy(pConsumer->user, subscribe->user, TSDB_USER_LEN);
51,630✔
155
    tstrncpy(pConsumer->fqdn, subscribe->fqdn, TSDB_FQDN_LEN);
51,630✔
156

157
    pConsumer->rebNewTopics = taosArrayDup(subscribe->topicNames, topicNameDup);
51,630✔
158
    if (pConsumer->rebNewTopics == NULL) {
51,630✔
159
      code = terrno;
×
160
      goto END;
×
161
    }
162
    pConsumer->assignedTopics = subscribe->topicNames;
51,630✔
163
    subscribe->topicNames = NULL;
51,630✔
164
  } else if (updateType == CONSUMER_UPDATE_SUB) {
82,507✔
165
    pConsumer->assignedTopics = subscribe->topicNames;
40,052✔
166
    subscribe->topicNames = NULL;
40,052✔
167
  }
168

169
  *ppConsumer = pConsumer;
227,605✔
170
  return 0;
227,605✔
171

172
END:
×
173
  tDeleteSMqConsumerObj(pConsumer);
×
174
  return code;
×
175
}
176

177
void tClearSMqConsumerObj(SMqConsumerObj *pConsumer) {
793,007✔
178
  if (pConsumer == NULL) return;
793,007✔
179
  taosArrayDestroyP(pConsumer->currentTopics, NULL);
455,483✔
180
  taosArrayDestroyP(pConsumer->rebNewTopics, NULL);
455,483✔
181
  taosArrayDestroyP(pConsumer->rebRemovedTopics, NULL);
455,483✔
182
  taosArrayDestroyP(pConsumer->assignedTopics, NULL);
455,483✔
183
}
184

185
void tDeleteSMqConsumerObj(SMqConsumerObj *pConsumer) {
565,129✔
186
  tClearSMqConsumerObj(pConsumer);
565,129✔
187
  taosMemoryFree(pConsumer);
565,129✔
188
}
565,129✔
189

190
int32_t tEncodeSMqConsumerObj(void **buf, const SMqConsumerObj *pConsumer) {
525,668✔
191
  int32_t tlen = 0;
525,668✔
192
  int32_t sz;
193
  tlen += taosEncodeFixedI64(buf, pConsumer->consumerId);
525,668✔
194
  tlen += taosEncodeString(buf, pConsumer->clientId);
525,668✔
195
  tlen += taosEncodeString(buf, pConsumer->cgroup);
525,668✔
196
  tlen += taosEncodeFixedI8(buf, pConsumer->updateType);
525,668✔
197
  tlen += taosEncodeFixedI32(buf, pConsumer->epoch);
525,668✔
198
  tlen += taosEncodeFixedI32(buf, pConsumer->status);
525,668✔
199

200
  tlen += taosEncodeFixedI32(buf, pConsumer->pid);
525,668✔
201
  tlen += taosEncodeSEpSet(buf, &pConsumer->ep);
525,668✔
202
  tlen += taosEncodeFixedI64(buf, pConsumer->createTime);
525,668✔
203
  tlen += taosEncodeFixedI64(buf, pConsumer->subscribeTime);
525,668✔
204
  tlen += taosEncodeFixedI64(buf, pConsumer->rebalanceTime);
525,668✔
205

206
  // current topics
207
  if (pConsumer->currentTopics) {
525,668✔
208
    sz = taosArrayGetSize(pConsumer->currentTopics);
71,924✔
209
    tlen += taosEncodeFixedI32(buf, sz);
71,924✔
210
    for (int32_t i = 0; i < sz; i++) {
113,814✔
211
      char *topic = taosArrayGetP(pConsumer->currentTopics, i);
41,890✔
212
      tlen += taosEncodeString(buf, topic);
41,890✔
213
    }
214
  } else {
215
    tlen += taosEncodeFixedI32(buf, 0);
453,744✔
216
  }
217

218
  // reb new topics
219
  if (pConsumer->rebNewTopics) {
525,668✔
220
    sz = taosArrayGetSize(pConsumer->rebNewTopics);
358,520✔
221
    tlen += taosEncodeFixedI32(buf, sz);
358,520✔
222
    for (int32_t i = 0; i < sz; i++) {
589,752✔
223
      char *topic = taosArrayGetP(pConsumer->rebNewTopics, i);
231,232✔
224
      tlen += taosEncodeString(buf, topic);
231,232✔
225
    }
226
  } else {
227
    tlen += taosEncodeFixedI32(buf, 0);
167,148✔
228
  }
229

230
  // reb removed topics
231
  if (pConsumer->rebRemovedTopics) {
525,668✔
232
    sz = taosArrayGetSize(pConsumer->rebRemovedTopics);
232,800✔
233
    tlen += taosEncodeFixedI32(buf, sz);
232,800✔
234
    for (int32_t i = 0; i < sz; i++) {
394,382✔
235
      char *topic = taosArrayGetP(pConsumer->rebRemovedTopics, i);
161,582✔
236
      tlen += taosEncodeString(buf, topic);
161,582✔
237
    }
238
  } else {
239
    tlen += taosEncodeFixedI32(buf, 0);
292,868✔
240
  }
241

242
  // lost topics
243
  if (pConsumer->assignedTopics) {
525,668✔
244
    sz = taosArrayGetSize(pConsumer->assignedTopics);
253,822✔
245
    tlen += taosEncodeFixedI32(buf, sz);
253,822✔
246
    for (int32_t i = 0; i < sz; i++) {
422,146✔
247
      char *topic = taosArrayGetP(pConsumer->assignedTopics, i);
168,324✔
248
      tlen += taosEncodeString(buf, topic);
168,324✔
249
    }
250
  } else {
251
    tlen += taosEncodeFixedI32(buf, 0);
271,846✔
252
  }
253

254
  tlen += taosEncodeFixedI8(buf, pConsumer->withTbName);
525,668✔
255
  tlen += taosEncodeFixedI8(buf, pConsumer->autoCommit);
525,668✔
256
  tlen += taosEncodeFixedI32(buf, pConsumer->autoCommitInterval);
525,668✔
257
  tlen += taosEncodeFixedI32(buf, pConsumer->resetOffsetCfg);
525,668✔
258
  tlen += taosEncodeFixedI32(buf, pConsumer->maxPollIntervalMs);
525,668✔
259
  tlen += taosEncodeFixedI32(buf, pConsumer->sessionTimeoutMs);
525,668✔
260
  tlen += taosEncodeString(buf, pConsumer->user);
525,668✔
261
  tlen += taosEncodeString(buf, pConsumer->fqdn);
525,668✔
262
  return tlen;
525,668✔
263
}
264

265
void *tDecodeSMqConsumerObj(const void *buf, SMqConsumerObj *pConsumer, int8_t sver) {
227,878✔
266
  int32_t sz;
223,426✔
267
  buf = taosDecodeFixedI64(buf, &pConsumer->consumerId);
227,878✔
268
  buf = taosDecodeStringTo(buf, pConsumer->clientId);
227,878✔
269
  buf = taosDecodeStringTo(buf, pConsumer->cgroup);
227,878✔
270
  buf = taosDecodeFixedI8(buf, &pConsumer->updateType);
227,878✔
271
  buf = taosDecodeFixedI32(buf, &pConsumer->epoch);
227,878✔
272
  buf = taosDecodeFixedI32(buf, &pConsumer->status);
227,878✔
273

274
  buf = taosDecodeFixedI32(buf, &pConsumer->pid);
227,878✔
275
  buf = taosDecodeSEpSet(buf, &pConsumer->ep);
227,878✔
276
  buf = taosDecodeFixedI64(buf, &pConsumer->createTime);
227,878✔
277
  buf = taosDecodeFixedI64(buf, &pConsumer->subscribeTime);
227,878✔
278
  buf = taosDecodeFixedI64(buf, &pConsumer->rebalanceTime);
455,756✔
279

280
  // current topics
281
  buf = taosDecodeFixedI32(buf, &sz);
227,878✔
282
  pConsumer->currentTopics = taosArrayInit(sz, sizeof(void *));
227,878✔
283
  if (pConsumer->currentTopics == NULL) {
227,878✔
284
    return NULL;
×
285
  }
286
  for (int32_t i = 0; i < sz; i++) {
228,596✔
287
    char *topic;
718✔
288
    buf = taosDecodeString(buf, &topic);
718✔
289
    if (taosArrayPush(pConsumer->currentTopics, &topic) == NULL) {
1,436✔
290
      return NULL;
×
291
    }
292
  }
293

294
  // reb new topics
295
  buf = taosDecodeFixedI32(buf, &sz);
227,878✔
296
  pConsumer->rebNewTopics = taosArrayInit(sz, sizeof(void *));
227,878✔
297
  for (int32_t i = 0; i < sz; i++) {
332,800✔
298
    char *topic;
103,014✔
299
    buf = taosDecodeString(buf, &topic);
104,922✔
300
    if (taosArrayPush(pConsumer->rebNewTopics, &topic) == NULL) {
209,844✔
301
      return NULL;
×
302
    }
303
  }
304

305
  // reb removed topics
306
  buf = taosDecodeFixedI32(buf, &sz);
227,878✔
307
  pConsumer->rebRemovedTopics = taosArrayInit(sz, sizeof(void *));
227,878✔
308
  for (int32_t i = 0; i < sz; i++) {
308,767✔
309
    char *topic;
78,981✔
310
    buf = taosDecodeString(buf, &topic);
80,889✔
311
    if (taosArrayPush(pConsumer->rebRemovedTopics, &topic) == NULL) {
161,778✔
312
      return NULL;
×
313
    }
314
  }
315

316
  // reb removed topics
317
  buf = taosDecodeFixedI32(buf, &sz);
227,878✔
318
  pConsumer->assignedTopics = taosArrayInit(sz, sizeof(void *));
227,878✔
319
  for (int32_t i = 0; i < sz; i++) {
281,033✔
320
    char *topic;
52,201✔
321
    buf = taosDecodeString(buf, &topic);
53,155✔
322
    if (taosArrayPush(pConsumer->assignedTopics, &topic) == NULL) {
106,310✔
323
      return NULL;
×
324
    }
325
  }
326

327
  if (sver > 1) {
227,878✔
328
    buf = taosDecodeFixedI8(buf, &pConsumer->withTbName);
227,878✔
329
    buf = taosDecodeFixedI8(buf, &pConsumer->autoCommit);
227,878✔
330
    buf = taosDecodeFixedI32(buf, &pConsumer->autoCommitInterval);
227,878✔
331
    buf = taosDecodeFixedI32(buf, &pConsumer->resetOffsetCfg);
455,756✔
332
  }
333
  if (sver > 2) {
227,878✔
334
    buf = taosDecodeFixedI32(buf, &pConsumer->maxPollIntervalMs);
227,878✔
335
    buf = taosDecodeFixedI32(buf, &pConsumer->sessionTimeoutMs);
227,878✔
336
    buf = taosDecodeStringTo(buf, pConsumer->user);
227,878✔
337
    buf = taosDecodeStringTo(buf, pConsumer->fqdn);
455,756✔
338
  } else {
339
    pConsumer->maxPollIntervalMs = DEFAULT_MAX_POLL_INTERVAL;
×
340
    pConsumer->sessionTimeoutMs = DEFAULT_SESSION_TIMEOUT;
×
341
  }
342

343
  return (void *)buf;
227,878✔
344
}
345

346
int32_t tEncodeOffRows(void **buf, SArray *offsetRows) {
436,510✔
347
  int32_t tlen = 0;
436,510✔
348
  int32_t szVgs = taosArrayGetSize(offsetRows);
436,510✔
349
  tlen += taosEncodeFixedI32(buf, szVgs);
436,510✔
350
  for (int32_t j = 0; j < szVgs; ++j) {
1,124,456✔
351
    OffsetRows *offRows = taosArrayGet(offsetRows, j);
687,946✔
352
    tlen += taosEncodeFixedI32(buf, offRows->vgId);
687,946✔
353
    tlen += taosEncodeFixedI64(buf, offRows->rows);
687,946✔
354
    tlen += taosEncodeFixedI8(buf, offRows->offset.type);
687,946✔
355
    if (offRows->offset.type == TMQ_OFFSET__SNAPSHOT_DATA || offRows->offset.type == TMQ_OFFSET__SNAPSHOT_META) {
687,946✔
356
      tlen += taosEncodeFixedI64(buf, offRows->offset.uid);
21,980✔
357
      tlen += taosEncodeFixedI64(buf, offRows->offset.ts);
43,960✔
358
    } else if (offRows->offset.type == TMQ_OFFSET__LOG) {
665,966✔
359
      tlen += taosEncodeFixedI64(buf, offRows->offset.version);
1,184,044✔
360
    } else {
361
      // do nothing
362
    }
363
    tlen += taosEncodeFixedI64(buf, offRows->ever);
1,375,892✔
364
  }
365

366
  return tlen;
436,510✔
367
}
368

369
int32_t tEncodeSMqConsumerEp(void **buf, const SMqConsumerEp *pConsumerEp) {
158,416✔
370
  int32_t tlen = 0;
158,416✔
371
  tlen += taosEncodeFixedI64(buf, pConsumerEp->consumerId);
158,416✔
372
  int32_t sz = taosArrayGetSize(pConsumerEp->vgs);
158,416✔
373
  tlen += taosEncodeFixedI32(buf, sz);
158,416✔
374
  for (int32_t i = 0; i < sz; i++) {
579,658✔
375
    void* data = taosArrayGet(pConsumerEp->vgs, i);
421,242✔
376
    tlen += tEncodeSMqVgEp(buf, data);
421,242✔
377
  }
378

379
  return tlen + tEncodeOffRows(buf, pConsumerEp->offsetRows);
158,416✔
380
}
381

382
void *tDecodeOffRows(const void *buf, SArray **offsetRows, int8_t sver) {
171,438✔
383
  int32_t szVgs = 0;
171,438✔
384
  buf = taosDecodeFixedI32(buf, &szVgs);
171,438✔
385
  if (szVgs > 0) {
171,438✔
386
    *offsetRows = taosArrayInit(szVgs, sizeof(OffsetRows));
73,312✔
387
    if (NULL == *offsetRows) return NULL;
73,312✔
388
    for (int32_t j = 0; j < szVgs; ++j) {
387,427✔
389
      OffsetRows *offRows = taosArrayReserve(*offsetRows, 1);
314,115✔
390
      buf = taosDecodeFixedI32(buf, &offRows->vgId);
314,115✔
391
      buf = taosDecodeFixedI64(buf, &offRows->rows);
314,115✔
392
      buf = taosDecodeFixedI8(buf, &offRows->offset.type);
314,115✔
393
      if (offRows->offset.type == TMQ_OFFSET__SNAPSHOT_DATA || offRows->offset.type == TMQ_OFFSET__SNAPSHOT_META) {
314,115✔
394
        buf = taosDecodeFixedI64(buf, &offRows->offset.uid);
9,986✔
395
        buf = taosDecodeFixedI64(buf, &offRows->offset.ts);
19,972✔
396
      } else if (offRows->offset.type == TMQ_OFFSET__LOG) {
304,129✔
397
        buf = taosDecodeFixedI64(buf, &offRows->offset.version);
541,748✔
398
      } else {
399
        // do nothing
400
      }
401
      if (sver > 2) {
314,115✔
402
        buf = taosDecodeFixedI64(buf, &offRows->ever);
628,230✔
403
      }
404
    }
405
  }
406
  return (void *)buf;
171,438✔
407
}
408

409
void *tDecodeSMqConsumerEp(const void *buf, SMqConsumerEp *pConsumerEp, int8_t sver) {
59,093✔
410
  buf = taosDecodeFixedI64(buf, &pConsumerEp->consumerId);
59,093✔
411
  int32_t sz = 0;
59,093✔
412
  buf = taosDecodeFixedI32(buf, &sz);
59,093✔
413
  pConsumerEp->vgs = taosArrayInit(sz, sizeof(SMqVgEp));
59,093✔
414
  if (pConsumerEp->vgs == NULL) {
59,093✔
415
    return NULL;
×
416
  }
417
  for (int32_t i = 0; i < sz; i++) {
249,173✔
418
    SMqVgEp* vgEp = taosArrayReserve(pConsumerEp->vgs, 1);
190,080✔
419
    if (vgEp != NULL)
190,080✔
420
      buf = tDecodeSMqVgEp(buf, vgEp, sver);
190,080✔
421
  }
422
  if (sver > 1) {
59,093✔
423
    buf = tDecodeOffRows(buf, &pConsumerEp->offsetRows, sver);
59,093✔
424
  }
425

426
  return (void *)buf;
59,093✔
427
}
428

429
int32_t tNewSubscribeObj(const char *key, SMqSubscribeObj **ppSub) {
37,187✔
430
  int32_t          code = 0;
37,187✔
431
  SMqSubscribeObj *pSubObj = taosMemoryCalloc(1, sizeof(SMqSubscribeObj));
37,187✔
432
  MND_TMQ_NULL_CHECK(pSubObj);
37,187✔
433
  *ppSub = pSubObj;
37,187✔
434
  
435
  (void)memcpy(pSubObj->key, key, TSDB_SUBSCRIBE_KEY_LEN);
37,187✔
436
  taosInitRWLatch(&pSubObj->lock);
37,187✔
437
  pSubObj->vgNum = 0;
37,187✔
438
  pSubObj->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
37,187✔
439
  MND_TMQ_NULL_CHECK(pSubObj->consumerHash);
37,187✔
440
  taosHashSetFreeFp(pSubObj->consumerHash, freeSMqConsumerEp);
37,187✔
441

442
  pSubObj->unassignedVgs = taosArrayInit(0, sizeof(SMqVgEp));
37,187✔
443
  MND_TMQ_NULL_CHECK(pSubObj->unassignedVgs);
37,187✔
444

445
END:
37,187✔
446
  return code;
37,187✔
447
}
448

449
int32_t tCloneSubscribeObj(const SMqSubscribeObj *pSub, SMqSubscribeObj **ppSub) {
57,700✔
450
  int32_t          code = 0;
57,700✔
451
  SMqSubscribeObj *pSubNew = taosMemoryMalloc(sizeof(SMqSubscribeObj));
57,700✔
452
  if (pSubNew == NULL) {
57,700✔
453
    code = terrno;
×
454
    goto END;
×
455
  }
456
  *ppSub = pSubNew;
57,700✔
457

458
  (void)memcpy(pSubNew->key, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);
57,700✔
459
  taosInitRWLatch(&pSubNew->lock);
57,700✔
460

461
  pSubNew->dbUid = pSub->dbUid;
57,700✔
462
  pSubNew->stbUid = pSub->stbUid;
57,700✔
463
  pSubNew->subType = pSub->subType;
57,700✔
464
  pSubNew->withMeta = pSub->withMeta;
57,700✔
465

466
  pSubNew->vgNum = pSub->vgNum;
57,700✔
467
  pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
57,700✔
468
  if (pSubNew->consumerHash == NULL) {
57,700✔
469
    code = terrno;
×
470
    goto END;
×
471
  }
472
  taosHashSetFreeFp(pSubNew->consumerHash, freeSMqConsumerEp);
57,700✔
473

474
  void          *pIter = NULL;
57,700✔
475
  SMqConsumerEp *pConsumerEp = NULL;
57,700✔
476
  while (1) {
50,412✔
477
    pIter = taosHashIterate(pSub->consumerHash, pIter);
108,112✔
478
    if (pIter == NULL) break;
108,112✔
479
    pConsumerEp = (SMqConsumerEp *)pIter;
50,412✔
480
    SMqConsumerEp newEp = {
51,366✔
481
        .consumerId = pConsumerEp->consumerId,
50,412✔
482
        .vgs = taosArrayDup(pConsumerEp->vgs, NULL),
50,412✔
483
    };
484
    if ((code = taosHashPut(pSubNew->consumerHash, &newEp.consumerId, sizeof(int64_t), &newEp,
50,412✔
485
                            sizeof(SMqConsumerEp))) != 0)
486
      goto END;
×
487
  }
488
  pSubNew->unassignedVgs = taosArrayDup(pSub->unassignedVgs, NULL);
57,700✔
489
  if (pSubNew->unassignedVgs == NULL) {
57,700✔
490
    code = terrno;
×
491
    goto END;
×
492
  }
493
  pSubNew->offsetRows = taosArrayDup(pSub->offsetRows, NULL);
57,700✔
494
  if (pSub->offsetRows != NULL && pSubNew->offsetRows == NULL) {
57,700✔
495
    code = terrno;
×
496
    goto END;
×
497
  }
498
  (void)memcpy(pSubNew->dbName, pSub->dbName, TSDB_DB_FNAME_LEN);
57,700✔
499
  pSubNew->qmsg = taosStrdup(pSub->qmsg);
57,700✔
500
  if (pSubNew->qmsg == NULL) {
57,700✔
501
    code = terrno;
×
502
    goto END;
×
503
  }
504

505
END:
57,700✔
506
  return code;
57,700✔
507
}
508

509
void tDeleteSubscribeObj(SMqSubscribeObj *pSub) {
207,232✔
510
  if (pSub == NULL) return;
207,232✔
511
  taosHashCleanup(pSub->consumerHash);
207,232✔
512
  pSub->consumerHash = NULL;
207,232✔
513
  taosArrayDestroy(pSub->unassignedVgs);
207,232✔
514
  pSub->unassignedVgs = NULL;
207,232✔
515
  taosMemoryFreeClear(pSub->qmsg);
207,232✔
516
  taosArrayDestroy(pSub->offsetRows);
207,232✔
517
  pSub->offsetRows = NULL;
207,232✔
518
}
519

520
int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {
278,094✔
521
  int32_t tlen = 0;
278,094✔
522
  tlen += taosEncodeString(buf, pSub->key);
278,094✔
523
  tlen += taosEncodeFixedI64(buf, pSub->dbUid);
278,094✔
524
  tlen += taosEncodeFixedI32(buf, pSub->vgNum);
278,094✔
525
  tlen += taosEncodeFixedI8(buf, pSub->subType);
278,094✔
526
  tlen += taosEncodeFixedI8(buf, pSub->withMeta);
278,094✔
527
  tlen += taosEncodeFixedI64(buf, pSub->stbUid);
278,094✔
528

529
  void   *pIter = NULL;
278,094✔
530
  int32_t sz = taosHashGetSize(pSub->consumerHash);
278,094✔
531
  tlen += taosEncodeFixedI32(buf, sz);
278,094✔
532

533
  int32_t cnt = 0;
278,094✔
534
  while (1) {
158,416✔
535
    pIter = taosHashIterate(pSub->consumerHash, pIter);
436,510✔
536
    if (pIter == NULL) break;
436,510✔
537
    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
158,416✔
538
    tlen += tEncodeSMqConsumerEp(buf, pConsumerEp);
158,416✔
539
    cnt++;
158,416✔
540
  }
541
  if (cnt != sz) return -1;
278,094✔
542
  int32_t len = taosArrayGetSize(pSub->unassignedVgs);
278,094✔
543
  tlen += taosEncodeFixedI32(buf, len);
278,094✔
544
  for (int32_t i = 0; i < len; i++) {
804,514✔
545
    void* data = taosArrayGet(pSub->unassignedVgs, i);
526,420✔
546
    tlen += tEncodeSMqVgEp(buf, data);
526,420✔
547
  }
548
  tlen += taosEncodeString(buf, pSub->dbName);
278,094✔
549

550
  tlen += tEncodeOffRows(buf, pSub->offsetRows);
278,094✔
551
  tlen += taosEncodeString(buf, pSub->qmsg);
278,094✔
552
  return tlen;
278,094✔
553
}
554

555
void *tDecodeSubscribeObj(const void *buf, SMqSubscribeObj *pSub, int8_t sver) {
112,345✔
556
  //
557
  buf = taosDecodeStringTo(buf, pSub->key);
112,345✔
558
  buf = taosDecodeFixedI64(buf, &pSub->dbUid);
112,345✔
559
  buf = taosDecodeFixedI32(buf, &pSub->vgNum);
112,345✔
560
  buf = taosDecodeFixedI8(buf, &pSub->subType);
112,345✔
561
  buf = taosDecodeFixedI8(buf, &pSub->withMeta);
112,345✔
562
  buf = taosDecodeFixedI64(buf, &pSub->stbUid);
114,253✔
563

564
  int32_t sz;
110,437✔
565
  buf = taosDecodeFixedI32(buf, &sz);
112,345✔
566

567
  pSub->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
112,345✔
568
  if (pSub->consumerHash == NULL) {
112,345✔
569
    return NULL;
×
570
  }
571
  taosHashSetFreeFp(pSub->consumerHash, freeSMqConsumerEp);
112,345✔
572

573
  for (int32_t i = 0; i < sz; i++) {
171,438✔
574
    SMqConsumerEp consumerEp = {0};
59,093✔
575
    buf = tDecodeSMqConsumerEp(buf, &consumerEp, sver);
59,093✔
576
    if (taosHashPut(pSub->consumerHash, &consumerEp.consumerId, sizeof(int64_t), &consumerEp, sizeof(SMqConsumerEp)) !=
59,093✔
577
        0)
578
      return NULL;
×
579
  }
580

581
  int32_t len = 0;
112,345✔
582
  buf = taosDecodeFixedI32(buf, &len);
112,345✔
583
  pSub->unassignedVgs = taosArrayInit(len, sizeof(SMqVgEp));
112,345✔
584
  if (pSub->unassignedVgs == NULL) {
112,345✔
585
    return NULL;
×
586
  }
587
  for (int32_t i = 0; i < len; i++) {
356,956✔
588
    SMqVgEp* vgEp = taosArrayReserve(pSub->unassignedVgs, 1);
244,611✔
589
    if (vgEp != NULL)
244,611✔
590
      buf = tDecodeSMqVgEp(buf, vgEp, sver);
244,611✔
591
  }
592

593
  buf = taosDecodeStringTo(buf, pSub->dbName);
112,345✔
594

595
  if (sver > 1) {
112,345✔
596
    buf = tDecodeOffRows(buf, &pSub->offsetRows, sver);
112,345✔
597
    buf = taosDecodeString(buf, &pSub->qmsg);
224,690✔
598
  } else {
599
    pSub->qmsg = taosStrdup("");
×
600
  }
601
  return (void *)buf;
112,345✔
602
}
603

604
int32_t mndInitConfigObj(SConfigItem *pItem, SConfigObj *pObj) {
33,174,608✔
605
  tstrncpy(pObj->name, pItem->name, CFG_NAME_MAX_LEN);
33,174,608✔
606
  pObj->dtype = pItem->dtype;
33,174,608✔
607
  switch (pItem->dtype) {
33,174,608✔
608
    case CFG_DTYPE_NONE:
×
609
      break;
×
610
    case CFG_DTYPE_BOOL:
6,572,328✔
611
      pObj->bval = pItem->bval;
6,572,328✔
612
      break;
6,572,328✔
613
    case CFG_DTYPE_INT32:
18,152,144✔
614
      pObj->i32 = pItem->i32;
18,152,144✔
615
      break;
18,152,144✔
616
    case CFG_DTYPE_INT64:
1,877,808✔
617
      pObj->i64 = pItem->i64;
1,877,808✔
618
      break;
1,877,808✔
619
    case CFG_DTYPE_FLOAT:
625,936✔
620
    case CFG_DTYPE_DOUBLE:
621
      pObj->fval = pItem->fval;
625,936✔
622
      break;
625,936✔
623
    case CFG_DTYPE_STRING:
5,946,392✔
624
    case CFG_DTYPE_DIR:
625
    case CFG_DTYPE_LOCALE:
626
    case CFG_DTYPE_CHARSET:
627
    case CFG_DTYPE_TIMEZONE:
628
      pObj->str = taosStrdup(pItem->str);
5,946,392✔
629
      if (pObj->str == NULL) {
5,946,392✔
630
        taosMemoryFree(pObj);
×
631
        return TSDB_CODE_OUT_OF_MEMORY;
×
632
      }
633
      break;
5,946,392✔
634
  }
635
  return TSDB_CODE_SUCCESS;
33,174,608✔
636
}
637

638
int32_t mndUpdateObj(SConfigObj *pObjNew, const char *name, char *value) {
18,178✔
639
  int32_t code = 0;
18,178✔
640
  switch (pObjNew->dtype) {
18,178✔
641
    case CFG_DTYPE_BOOL: {
3,082✔
642
      bool tmp = false;
3,082✔
643
      if (strcasecmp(value, "true") == 0) {
3,082✔
644
        tmp = true;
×
645
      }
646
      if (taosStr2Int32(value, NULL, 10) > 0) {
3,082✔
647
        tmp = true;
1,065✔
648
      }
649
      pObjNew->bval = tmp;
3,082✔
650
      break;
3,082✔
651
    }
652
    case CFG_DTYPE_INT32: {
13,739✔
653
      int32_t ival;
13,739✔
654
      TAOS_CHECK_RETURN(taosStrHumanToInt32(value, &ival));
13,739✔
655
      pObjNew->i32 = ival;
13,739✔
656
      break;
13,739✔
657
    }
658
    case CFG_DTYPE_INT64: {
1,083✔
659
      int64_t ival;
1,083✔
660
      TAOS_CHECK_RETURN(taosStrHumanToInt64(value, &ival));
1,083✔
661
      pObjNew->i64 = ival;
1,083✔
662
      break;
1,083✔
663
    }
664
    case CFG_DTYPE_FLOAT:
×
665
    case CFG_DTYPE_DOUBLE: {
666
      float dval = 0;
×
667
      TAOS_CHECK_RETURN(parseCfgReal(value, &dval));
×
668
      pObjNew->fval = dval;
×
669
      break;
×
670
    }
671
    case CFG_DTYPE_DIR:
274✔
672
    case CFG_DTYPE_TIMEZONE:
673
    case CFG_DTYPE_CHARSET:
674
    case CFG_DTYPE_LOCALE:
675
    case CFG_DTYPE_STRING: {
676
      pObjNew->str = taosStrdup(value);
274✔
677
      if (pObjNew->str == NULL) {
274✔
678
        code = terrno;
×
679
        return code;
×
680
      }
681
      break;
274✔
682
    }
683
    case CFG_DTYPE_NONE:
×
684
      break;
×
685
    default:
×
686
      code = TSDB_CODE_INVALID_CFG;
×
687
      break;
×
688
  }
689
  return code;
18,178✔
690
}
691

692
SConfigObj mndInitConfigVersion() {
312,968✔
693
  SConfigObj obj;
311,399✔
694
  memset(&obj, 0, sizeof(SConfigObj));
312,968✔
695

696
  tstrncpy(obj.name, "tsmmConfigVersion", CFG_NAME_MAX_LEN);
312,968✔
697
  obj.dtype = CFG_DTYPE_INT32;
312,968✔
698
  obj.i32 = 0;
312,968✔
699
  return obj;
312,968✔
700
}
701

702
int32_t tEncodeSConfigObj(SEncoder *pEncoder, const SConfigObj *pObj) {
255,607,190✔
703
  TAOS_CHECK_RETURN(tStartEncode(pEncoder));
255,607,190✔
704
  TAOS_CHECK_RETURN(tEncodeCStr(pEncoder, pObj->name));
511,214,380✔
705

706
  TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pObj->dtype));
511,214,380✔
707
  switch (pObj->dtype) {
255,607,190✔
708
    case CFG_DTYPE_BOOL:
50,150,642✔
709
      TAOS_CHECK_RETURN(tEncodeI8(pEncoder, pObj->bval));
100,301,284✔
710
      break;
50,150,642✔
711
    case CFG_DTYPE_INT32:
140,989,106✔
712
      TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pObj->i32));
281,978,212✔
713
      break;
140,989,106✔
714
    case CFG_DTYPE_INT64:
14,329,224✔
715
      TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pObj->i64));
28,658,448✔
716
      break;
14,329,224✔
717
    case CFG_DTYPE_FLOAT:
4,774,964✔
718
    case CFG_DTYPE_DOUBLE:
719
      TAOS_CHECK_RETURN(tEncodeFloat(pEncoder, pObj->fval));
9,549,928✔
720
      break;
4,774,964✔
721
    case CFG_DTYPE_STRING:
45,363,254✔
722
    case CFG_DTYPE_DIR:
723
    case CFG_DTYPE_LOCALE:
724
    case CFG_DTYPE_CHARSET:
725
    case CFG_DTYPE_TIMEZONE:
726
      if (pObj->str != NULL) {
45,363,254✔
727
        TAOS_CHECK_RETURN(tEncodeCStr(pEncoder, pObj->str));
90,726,508✔
728
      } else {
729
        TAOS_CHECK_RETURN(tEncodeCStr(pEncoder, ""));
×
730
      }
731
      break;
45,363,254✔
732
    default:
×
733
      break;
×
734
  }
735
  tEndEncode(pEncoder);
255,607,190✔
736
  return pEncoder->pos;
255,607,190✔
737
}
738

739
int32_t tDecodeSConfigObj(SDecoder *pDecoder, SConfigObj *pObj) {
51,844,712✔
740
  TAOS_CHECK_RETURN(tStartDecode(pDecoder));
51,844,712✔
741
  TAOS_CHECK_RETURN(tDecodeCStrTo(pDecoder, pObj->name));
51,844,712✔
742
  TAOS_CHECK_RETURN(tDecodeI32(pDecoder, (int32_t *)&pObj->dtype));
103,689,424✔
743
  switch (pObj->dtype) {
51,844,712✔
744
    case CFG_DTYPE_NONE:
×
745
      break;
×
746
    case CFG_DTYPE_BOOL:
10,158,573✔
747
      TAOS_CHECK_RETURN(tDecodeBool(pDecoder, &pObj->bval));
10,158,573✔
748
      break;
10,158,573✔
749
    case CFG_DTYPE_INT32:
28,635,128✔
750
      TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pObj->i32));
57,270,256✔
751
      break;
28,635,128✔
752
    case CFG_DTYPE_INT64:
2,904,070✔
753
      TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pObj->i64));
5,808,140✔
754
      break;
2,904,070✔
755
    case CFG_DTYPE_FLOAT:
968,030✔
756
    case CFG_DTYPE_DOUBLE:
757
      TAOS_CHECK_RETURN(tDecodeFloat(pDecoder, &pObj->fval));
1,936,060✔
758
      break;
968,030✔
759
    case CFG_DTYPE_STRING:
9,178,911✔
760
    case CFG_DTYPE_DIR:
761
    case CFG_DTYPE_LOCALE:
762
    case CFG_DTYPE_CHARSET:
763
    case CFG_DTYPE_TIMEZONE:
764
      TAOS_CHECK_RETURN(tDecodeCStrAlloc(pDecoder, &pObj->str));
18,357,822✔
765
      break;
9,178,911✔
766
  }
767
  tEndDecode(pDecoder);
51,844,712✔
768
  TAOS_RETURN(TSDB_CODE_SUCCESS);
51,844,712✔
769
}
770

771
void tFreeSConfigObj(SConfigObj *obj) {
85,311,613✔
772
  if (obj == NULL) {
85,311,613✔
773
    return;
×
774
  }
775
  if (obj->dtype == CFG_DTYPE_STRING || obj->dtype == CFG_DTYPE_DIR || obj->dtype == CFG_DTYPE_LOCALE ||
85,311,613✔
776
      obj->dtype == CFG_DTYPE_CHARSET || obj->dtype == CFG_DTYPE_TIMEZONE) {
71,787,405✔
777
    taosMemoryFree(obj->str);
15,115,450✔
778
  }
779
}
780

781
// SMqSubActionLogEntry *tCloneSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {
782
//   SMqSubActionLogEntry *pEntryNew = taosMemoryMalloc(sizeof(SMqSubActionLogEntry));
783
//   if (pEntryNew == NULL) return NULL;
784
//   pEntryNew->epoch = pEntry->epoch;
785
//   pEntryNew->consumers = taosArrayDup(pEntry->consumers, (__array_item_dup_fn_t)tCloneSMqConsumerEp);
786
//   return pEntryNew;
787
// }
788
//
789
// void tDeleteSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {
790
//   taosArrayDestroyEx(pEntry->consumers, (FDelete)tDeleteSMqConsumerEp);
791
// }
792

793
// int32_t tEncodeSMqSubActionLogEntry(void **buf, const SMqSubActionLogEntry *pEntry) {
794
//   int32_t tlen = 0;
795
//   tlen += taosEncodeFixedI32(buf, pEntry->epoch);
796
//   tlen += taosEncodeArray(buf, pEntry->consumers, (FEncode)tEncodeSMqSubActionLogEntry);
797
//   return tlen;
798
// }
799
//
800
// void *tDecodeSMqSubActionLogEntry(const void *buf, SMqSubActionLogEntry *pEntry) {
801
//   buf = taosDecodeFixedI32(buf, &pEntry->epoch);
802
//   buf = taosDecodeArray(buf, &pEntry->consumers, (FDecode)tDecodeSMqSubActionLogEntry, sizeof(SMqSubActionLogEntry));
803
//   return (void *)buf;
804
// }
805

806
// SMqSubActionLogObj *tCloneSMqSubActionLogObj(SMqSubActionLogObj *pLog) {
807
//   SMqSubActionLogObj *pLogNew = taosMemoryMalloc(sizeof(SMqSubActionLogObj));
808
//   if (pLogNew == NULL) return pLogNew;
809
//   memcpy(pLogNew->key, pLog->key, TSDB_SUBSCRIBE_KEY_LEN);
810
//   pLogNew->logs = taosArrayDup(pLog->logs, (__array_item_dup_fn_t)tCloneSMqConsumerEp);
811
//   return pLogNew;
812
// }
813
//
814
// void tDeleteSMqSubActionLogObj(SMqSubActionLogObj *pLog) {
815
//   taosArrayDestroyEx(pLog->logs, (FDelete)tDeleteSMqConsumerEp);
816
// }
817

818
// int32_t tEncodeSMqSubActionLogObj(void **buf, const SMqSubActionLogObj *pLog) {
819
//   int32_t tlen = 0;
820
//   tlen += taosEncodeString(buf, pLog->key);
821
//   tlen += taosEncodeArray(buf, pLog->logs, (FEncode)tEncodeSMqSubActionLogEntry);
822
//   return tlen;
823
// }
824
//
825
// void *tDecodeSMqSubActionLogObj(const void *buf, SMqSubActionLogObj *pLog) {
826
//   buf = taosDecodeStringTo(buf, pLog->key);
827
//   buf = taosDecodeArray(buf, &pLog->logs, (FDecode)tDecodeSMqSubActionLogEntry, sizeof(SMqSubActionLogEntry));
828
//   return (void *)buf;
829
// }
830
//
831
// int32_t tEncodeSMqOffsetObj(void **buf, const SMqOffsetObj *pOffset) {
832
//   int32_t tlen = 0;
833
//   tlen += taosEncodeString(buf, pOffset->key);
834
//   tlen += taosEncodeFixedI64(buf, pOffset->offset);
835
//   return tlen;
836
// }
837
//
838
// void *tDecodeSMqOffsetObj(void *buf, SMqOffsetObj *pOffset) {
839
//   buf = taosDecodeStringTo(buf, pOffset->key);
840
//   buf = taosDecodeFixedI64(buf, &pOffset->offset);
841
//   return buf;
842
// }
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc