tmq.c 37.4 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
/*
 * 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 "clientInt.h"
#include "clientLog.h"
#include "parser.h"
H
Haojun Liao 已提交
19
#include "tdatablock.h"
20 21 22
#include "tdef.h"
#include "tglobal.h"
#include "tmsgtype.h"
X
Xiaoyu Wang 已提交
23
#include "tqueue.h"
24
#include "tref.h"
L
Liu Jicong 已提交
25 26
#include "ttimer.h"

L
Liu Jicong 已提交
27 28 29 30 31 32 33 34
int32_t tmqAskEp(tmq_t* tmq, bool async);

typedef struct {
  int8_t inited;
  tmr_h  timer;
} SMqMgmt;

static SMqMgmt tmqMgmt = {0};
35

L
Liu Jicong 已提交
36 37 38 39 40 41
typedef struct {
  int8_t  tmqRspType;
  int32_t epoch;
} SMqRspWrapper;

typedef struct {
L
Liu Jicong 已提交
42 43 44
  int8_t      tmqRspType;
  int32_t     epoch;
  SMqAskEpRsp msg;
L
Liu Jicong 已提交
45 46
} SMqAskEpRspWrapper;

L
Liu Jicong 已提交
47
struct tmq_list_t {
L
Liu Jicong 已提交
48
  SArray container;
L
Liu Jicong 已提交
49
};
L
Liu Jicong 已提交
50

L
Liu Jicong 已提交
51
struct tmq_topic_vgroup_t {
L
Liu Jicong 已提交
52
  SMqOffset offset;
L
Liu Jicong 已提交
53 54 55
};

struct tmq_topic_vgroup_list_t {
L
Liu Jicong 已提交
56
  SArray container;  // SArray<tmq_topic_vgroup_t*>
L
Liu Jicong 已提交
57 58 59
};

struct tmq_conf_t {
L
Liu Jicong 已提交
60
  char           clientId[256];
L
Liu Jicong 已提交
61
  char           groupId[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
62
  int8_t         autoCommit;
L
Liu Jicong 已提交
63
  int8_t         resetOffset;
L
Liu Jicong 已提交
64
  uint16_t       port;
L
Liu Jicong 已提交
65
  int32_t        autoCommitInterval;
L
Liu Jicong 已提交
66 67 68 69
  char*          ip;
  char*          user;
  char*          pass;
  char*          db;
70
  tmq_commit_cb* commitCb;
L
Liu Jicong 已提交
71 72 73
};

struct tmq_t {
L
Liu Jicong 已提交
74
  // conf
L
Liu Jicong 已提交
75 76 77
  char           groupId[TSDB_CGROUP_LEN];
  char           clientId[256];
  int8_t         autoCommit;
L
Liu Jicong 已提交
78
  int32_t        autoCommitInterval;
L
Liu Jicong 已提交
79
  int32_t        resetOffsetCfg;
L
Liu Jicong 已提交
80
  int64_t        consumerId;
L
Liu Jicong 已提交
81
  tmq_commit_cb* commit_cb;
L
Liu Jicong 已提交
82 83 84 85

  // status
  int8_t  status;
  int32_t epoch;
L
Liu Jicong 已提交
86 87
#if 0
  int8_t  epStatus;
L
Liu Jicong 已提交
88
  int32_t epSkipCnt;
L
Liu Jicong 已提交
89
#endif
L
Liu Jicong 已提交
90 91
  int64_t pollCnt;

L
Liu Jicong 已提交
92 93 94 95 96
  // timer
  tmr_h hbTimer;
  tmr_h reportTimer;
  tmr_h commitTimer;

L
Liu Jicong 已提交
97 98 99 100
  // connection
  STscObj* pTscObj;

  // container
L
Liu Jicong 已提交
101
  SArray*     clientTopics;  // SArray<SMqClientTopic>
L
Liu Jicong 已提交
102
  STaosQueue* mqueue;        // queue of rsp
L
Liu Jicong 已提交
103
  STaosQall*  qall;
L
Liu Jicong 已提交
104 105 106 107
  STaosQueue* delayedTask;  // delayed task queue for heartbeat and auto commit

  // ctl
  tsem_t rspSem;
L
Liu Jicong 已提交
108 109
};

X
Xiaoyu Wang 已提交
110 111 112 113 114 115 116 117
enum {
  TMQ_VG_STATUS__IDLE = 0,
  TMQ_VG_STATUS__WAIT,
};

enum {
  TMQ_CONSUMER_STATUS__INIT = 0,
  TMQ_CONSUMER_STATUS__READY,
118
  TMQ_CONSUMER_STATUS__NO_TOPIC,
L
Liu Jicong 已提交
119 120
};

L
Liu Jicong 已提交
121 122 123 124 125 126
enum {
  TMQ_DELAYED_TASK__HB = 1,
  TMQ_DELAYED_TASK__REPORT,
  TMQ_DELAYED_TASK__COMMIT,
};

L
Liu Jicong 已提交
127
typedef struct {
128 129 130 131
  // statistics
  int64_t pollCnt;
  // offset
  int64_t currentOffset;
L
Liu Jicong 已提交
132
  // connection info
133
  int32_t vgId;
X
Xiaoyu Wang 已提交
134
  int32_t vgStatus;
L
Liu Jicong 已提交
135
  int32_t vgSkipCnt;
136 137 138
  SEpSet  epSet;
} SMqClientVg;

L
Liu Jicong 已提交
139
typedef struct {
140
  // subscribe info
L
Liu Jicong 已提交
141 142 143 144
  char* topicName;

  SArray* vgs;  // SArray<SMqClientVg>

L
Liu Jicong 已提交
145 146
  int8_t         isSchemaAdaptive;
  SSchemaWrapper schema;
147 148
} SMqClientTopic;

L
Liu Jicong 已提交
149 150 151 152 153
typedef struct {
  int8_t          tmqRspType;
  int32_t         epoch;
  SMqClientVg*    vgHandle;
  SMqClientTopic* topicHandle;
L
Liu Jicong 已提交
154
  SMqDataBlkRsp   msg;
L
Liu Jicong 已提交
155 156
} SMqPollRspWrapper;

L
Liu Jicong 已提交
157
typedef struct {
L
Liu Jicong 已提交
158 159 160 161
  tmq_t*         tmq;
  tsem_t         rspSem;
  tmq_resp_err_t rspErr;
} SMqSubscribeCbParam;
L
Liu Jicong 已提交
162

L
Liu Jicong 已提交
163
typedef struct {
164
  tmq_t*  tmq;
L
Liu Jicong 已提交
165
  int32_t code;
L
Liu Jicong 已提交
166
  int32_t async;
X
Xiaoyu Wang 已提交
167
  tsem_t  rspSem;
168 169
} SMqAskEpCbParam;

L
Liu Jicong 已提交
170
typedef struct {
L
Liu Jicong 已提交
171 172
  tmq_t*          tmq;
  SMqClientVg*    pVg;
L
Liu Jicong 已提交
173
  SMqClientTopic* pTopic;
L
Liu Jicong 已提交
174
  int32_t         epoch;
L
Liu Jicong 已提交
175
  int32_t         vgId;
L
Liu Jicong 已提交
176
  tsem_t          rspSem;
X
Xiaoyu Wang 已提交
177
} SMqPollCbParam;
178

L
Liu Jicong 已提交
179
typedef struct {
L
Liu Jicong 已提交
180
  tmq_t*         tmq;
L
Liu Jicong 已提交
181
  int32_t        async;
L
Liu Jicong 已提交
182 183
  tsem_t         rspSem;
  tmq_resp_err_t rspErr;
L
Liu Jicong 已提交
184
} SMqCommitCbParam;
L
Liu Jicong 已提交
185

186
tmq_conf_t* tmq_conf_new() {
wafwerar's avatar
wafwerar 已提交
187
  tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t));
L
Liu Jicong 已提交
188
  conf->autoCommit = false;
L
Liu Jicong 已提交
189
  conf->autoCommitInterval = 5000;
X
Xiaoyu Wang 已提交
190
  conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
191 192 193
  return conf;
}

L
Liu Jicong 已提交
194
void tmq_conf_destroy(tmq_conf_t* conf) {
wafwerar's avatar
wafwerar 已提交
195
  if (conf) taosMemoryFree(conf);
L
Liu Jicong 已提交
196 197 198
}

tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value) {
199 200
  if (strcmp(key, "group.id") == 0) {
    strcpy(conf->groupId, value);
L
Liu Jicong 已提交
201
    return TMQ_CONF_OK;
202
  }
L
Liu Jicong 已提交
203

204 205
  if (strcmp(key, "client.id") == 0) {
    strcpy(conf->clientId, value);
L
Liu Jicong 已提交
206 207
    return TMQ_CONF_OK;
  }
L
Liu Jicong 已提交
208

L
Liu Jicong 已提交
209 210
  if (strcmp(key, "enable.auto.commit") == 0) {
    if (strcmp(value, "true") == 0) {
L
Liu Jicong 已提交
211
      conf->autoCommit = true;
L
Liu Jicong 已提交
212 213
      return TMQ_CONF_OK;
    } else if (strcmp(value, "false") == 0) {
L
Liu Jicong 已提交
214
      conf->autoCommit = false;
L
Liu Jicong 已提交
215 216 217 218
      return TMQ_CONF_OK;
    } else {
      return TMQ_CONF_INVALID;
    }
219
  }
L
Liu Jicong 已提交
220

L
Liu Jicong 已提交
221 222 223 224 225
  if (strcmp(key, "auto.commit.interval.ms") == 0) {
    conf->autoCommitInterval = atoi(value);
    return TMQ_CONF_OK;
  }

L
Liu Jicong 已提交
226 227 228 229 230 231 232 233 234 235 236 237 238 239
  if (strcmp(key, "auto.offset.reset") == 0) {
    if (strcmp(value, "none") == 0) {
      conf->resetOffset = TMQ_CONF__RESET_OFFSET__NONE;
      return TMQ_CONF_OK;
    } else if (strcmp(value, "earliest") == 0) {
      conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
      return TMQ_CONF_OK;
    } else if (strcmp(value, "latest") == 0) {
      conf->resetOffset = TMQ_CONF__RESET_OFFSET__LATEST;
      return TMQ_CONF_OK;
    } else {
      return TMQ_CONF_INVALID;
    }
  }
L
Liu Jicong 已提交
240

L
Liu Jicong 已提交
241
  if (strcmp(key, "td.connect.ip") == 0) {
L
Liu Jicong 已提交
242 243 244
    conf->ip = strdup(value);
    return TMQ_CONF_OK;
  }
L
Liu Jicong 已提交
245
  if (strcmp(key, "td.connect.user") == 0) {
L
Liu Jicong 已提交
246 247 248
    conf->user = strdup(value);
    return TMQ_CONF_OK;
  }
L
Liu Jicong 已提交
249
  if (strcmp(key, "td.connect.pass") == 0) {
L
Liu Jicong 已提交
250 251 252
    conf->pass = strdup(value);
    return TMQ_CONF_OK;
  }
L
Liu Jicong 已提交
253
  if (strcmp(key, "td.connect.port") == 0) {
L
Liu Jicong 已提交
254 255 256
    conf->port = atoi(value);
    return TMQ_CONF_OK;
  }
L
Liu Jicong 已提交
257
  if (strcmp(key, "td.connect.db") == 0) {
L
Liu Jicong 已提交
258 259 260 261
    conf->db = strdup(value);
    return TMQ_CONF_OK;
  }

L
Liu Jicong 已提交
262
  return TMQ_CONF_UNKNOWN;
263 264 265
}

tmq_list_t* tmq_list_new() {
L
Liu Jicong 已提交
266 267
  //
  return (tmq_list_t*)taosArrayInit(0, sizeof(void*));
268 269
}

L
Liu Jicong 已提交
270 271 272
int32_t tmq_list_append(tmq_list_t* list, const char* src) {
  SArray* container = &list->container;
  char*   topic = strdup(src);
L
fix  
Liu Jicong 已提交
273
  if (taosArrayPush(container, &topic) == NULL) return -1;
274 275 276
  return 0;
}

L
Liu Jicong 已提交
277
void tmq_list_destroy(tmq_list_t* list) {
L
Liu Jicong 已提交
278
  SArray* container = &list->container;
L
Liu Jicong 已提交
279
  taosArrayDestroyP(container, taosMemoryFree);
L
Liu Jicong 已提交
280 281
}

L
Liu Jicong 已提交
282 283 284 285 286 287 288 289 290 291
int32_t tmq_list_get_size(const tmq_list_t* list) {
  const SArray* container = &list->container;
  return taosArrayGetSize(container);
}

char** tmq_list_to_c_array(const tmq_list_t* list) {
  const SArray* container = &list->container;
  return container->pData;
}

L
Liu Jicong 已提交
292 293 294 295
static int32_t tmqMakeTopicVgKey(char* dst, const char* topicName, int32_t vg) {
  return sprintf(dst, "%s:%d", topicName, vg);
}

L
Liu Jicong 已提交
296 297 298 299 300
void tmqAssignDelayedHbTask(void* param, void* tmrId) {
  tmq_t*  tmq = (tmq_t*)param;
  int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t));
  *pTaskType = TMQ_DELAYED_TASK__HB;
  taosWriteQitem(tmq->delayedTask, pTaskType);
301
  tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
302 303 304 305 306 307 308
}

void tmqAssignDelayedCommitTask(void* param, void* tmrId) {
  tmq_t*  tmq = (tmq_t*)param;
  int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t));
  *pTaskType = TMQ_DELAYED_TASK__COMMIT;
  taosWriteQitem(tmq->delayedTask, pTaskType);
309
  tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
310 311 312 313 314 315 316
}

void tmqAssignDelayedReportTask(void* param, void* tmrId) {
  tmq_t*  tmq = (tmq_t*)param;
  int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t));
  *pTaskType = TMQ_DELAYED_TASK__REPORT;
  taosWriteQitem(tmq->delayedTask, pTaskType);
317
  tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
318 319 320 321 322 323 324 325 326 327 328
}

int32_t tmqHandleAllDelayedTask(tmq_t* tmq) {
  STaosQall* qall = taosAllocateQall();
  taosReadAllQitems(tmq->delayedTask, qall);
  while (1) {
    int8_t* pTaskType = NULL;
    taosGetQitem(qall, (void**)&pTaskType);
    if (pTaskType == NULL) break;

    if (*pTaskType == TMQ_DELAYED_TASK__HB) {
L
Liu Jicong 已提交
329
      tmqAskEp(tmq, true);
L
Liu Jicong 已提交
330 331 332 333 334 335 336 337 338 339 340 341 342
      taosTmrReset(tmqAssignDelayedHbTask, 1000, tmq, tmqMgmt.timer, &tmq->hbTimer);
    } else if (*pTaskType == TMQ_DELAYED_TASK__COMMIT) {
      tmq_commit(tmq, NULL, true);
      taosTmrReset(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, tmq, tmqMgmt.timer, &tmq->commitTimer);
    } else if (*pTaskType == TMQ_DELAYED_TASK__REPORT) {
    } else {
      ASSERT(0);
    }
  }
  taosFreeQall(qall);
  return 0;
}

L
Liu Jicong 已提交
343
void tmqClearUnhandleMsg(tmq_t* tmq) {
L
Liu Jicong 已提交
344
  SMqRspWrapper* msg = NULL;
L
Liu Jicong 已提交
345 346 347 348 349 350 351 352
  while (1) {
    taosGetQitem(tmq->qall, (void**)&msg);
    if (msg)
      taosFreeQitem(msg);
    else
      break;
  }

L
fix  
Liu Jicong 已提交
353
  msg = NULL;
L
Liu Jicong 已提交
354 355 356 357 358 359 360 361 362 363
  taosReadAllQitems(tmq->mqueue, tmq->qall);
  while (1) {
    taosGetQitem(tmq->qall, (void**)&msg);
    if (msg)
      taosFreeQitem(msg);
    else
      break;
  }
}

L
Liu Jicong 已提交
364 365 366
int32_t tmqSubscribeCb(void* param, const SDataBuf* pMsg, int32_t code) {
  SMqSubscribeCbParam* pParam = (SMqSubscribeCbParam*)param;
  pParam->rspErr = code;
L
Liu Jicong 已提交
367
  tmq_t* tmq = pParam->tmq;
L
Liu Jicong 已提交
368 369 370
  tsem_post(&pParam->rspSem);
  return 0;
}
371

L
Liu Jicong 已提交
372
int32_t tmqCommitCb(void* param, const SDataBuf* pMsg, int32_t code) {
L
Liu Jicong 已提交
373
  SMqCommitCbParam* pParam = (SMqCommitCbParam*)param;
L
Liu Jicong 已提交
374
  pParam->rspErr = code == 0 ? TMQ_RESP_ERR__SUCCESS : TMQ_RESP_ERR__FAIL;
L
Liu Jicong 已提交
375
  if (pParam->tmq->commit_cb) {
L
Liu Jicong 已提交
376
    pParam->tmq->commit_cb(pParam->tmq, pParam->rspErr, NULL);
L
Liu Jicong 已提交
377
  }
L
fix  
Liu Jicong 已提交
378
  if (!pParam->async) tsem_post(&pParam->rspSem);
L
Liu Jicong 已提交
379 380 381
  return 0;
}

X
Xiaoyu Wang 已提交
382 383 384 385 386 387
tmq_resp_err_t tmq_subscription(tmq_t* tmq, tmq_list_t** topics) {
  if (*topics == NULL) {
    *topics = tmq_list_new();
  }
  for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
    SMqClientTopic* topic = taosArrayGetP(tmq->clientTopics, i);
L
Liu Jicong 已提交
388
    tmq_list_append(*topics, topic->topicName);
X
Xiaoyu Wang 已提交
389 390 391 392 393
  }
  return TMQ_RESP_ERR__SUCCESS;
}

tmq_resp_err_t tmq_unsubscribe(tmq_t* tmq) {
L
Liu Jicong 已提交
394 395 396 397
  tmq_list_t*    lst = tmq_list_new();
  tmq_resp_err_t rsp = tmq_subscribe(tmq, lst);
  tmq_list_destroy(lst);
  return rsp;
X
Xiaoyu Wang 已提交
398 399
}

L
Liu Jicong 已提交
400
#if 0
401
tmq_t* tmq_consumer_new(void* conn, tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
wafwerar's avatar
wafwerar 已提交
402
  tmq_t* pTmq = taosMemoryCalloc(sizeof(tmq_t), 1);
403 404 405 406 407 408
  if (pTmq == NULL) {
    return NULL;
  }
  pTmq->pTscObj = (STscObj*)conn;
  pTmq->status = 0;
  pTmq->pollCnt = 0;
L
Liu Jicong 已提交
409
  pTmq->epoch = 0;
L
fix txn  
Liu Jicong 已提交
410
  pTmq->epStatus = 0;
L
temp  
Liu Jicong 已提交
411
  pTmq->epSkipCnt = 0;
L
Liu Jicong 已提交
412
  // set conf
413 414
  strcpy(pTmq->clientId, conf->clientId);
  strcpy(pTmq->groupId, conf->groupId);
L
Liu Jicong 已提交
415
  pTmq->autoCommit = conf->autoCommit;
416
  pTmq->commit_cb = conf->commit_cb;
L
Liu Jicong 已提交
417
  pTmq->resetOffsetCfg = conf->resetOffset;
L
Liu Jicong 已提交
418

L
Liu Jicong 已提交
419 420 421 422 423 424 425 426 427 428
  pTmq->consumerId = generateRequestId() & (((uint64_t)-1) >> 1);
  pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic));
  if (pTmq->clientTopics == NULL) {
    taosMemoryFree(pTmq);
    return NULL;
  }

  pTmq->mqueue = taosOpenQueue();
  pTmq->qall = taosAllocateQall();

L
Liu Jicong 已提交
429 430
  tsem_init(&pTmq->rspSem, 0, 0);

L
Liu Jicong 已提交
431 432
  return pTmq;
}
L
Liu Jicong 已提交
433
#endif
L
Liu Jicong 已提交
434

L
Liu Jicong 已提交
435
tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
L
Liu Jicong 已提交
436 437 438 439 440 441 442 443 444 445 446
  // init timer
  int8_t inited = atomic_val_compare_exchange_8(&tmqMgmt.inited, 0, 1);
  if (inited == 0) {
    tmqMgmt.timer = taosTmrInit(1000, 100, 360000, "TMQ");
    if (tmqMgmt.timer == NULL) {
      atomic_store_8(&tmqMgmt.inited, 0);
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return NULL;
    }
  }

L
Liu Jicong 已提交
447 448 449 450
  tmq_t* pTmq = taosMemoryCalloc(1, sizeof(tmq_t));
  if (pTmq == NULL) {
    return NULL;
  }
L
Liu Jicong 已提交
451

L
Liu Jicong 已提交
452 453 454 455 456
  const char* user = conf->user == NULL ? TSDB_DEFAULT_USER : conf->user;
  const char* pass = conf->pass == NULL ? TSDB_DEFAULT_PASS : conf->pass;

  ASSERT(user);
  ASSERT(pass);
L
Liu Jicong 已提交
457
  ASSERT(conf->groupId[0]);
L
Liu Jicong 已提交
458

L
Liu Jicong 已提交
459 460 461 462 463 464 465 466
  pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic));
  pTmq->mqueue = taosOpenQueue();
  pTmq->qall = taosAllocateQall();
  pTmq->delayedTask = taosOpenQueue();

  if (pTmq->clientTopics == NULL || pTmq->mqueue == NULL || pTmq->qall == NULL || pTmq->delayedTask == NULL) {
    goto FAIL;
  }
L
Liu Jicong 已提交
467

L
Liu Jicong 已提交
468 469
  // init status
  pTmq->status = TMQ_CONSUMER_STATUS__INIT;
L
Liu Jicong 已提交
470 471
  pTmq->pollCnt = 0;
  pTmq->epoch = 0;
L
Liu Jicong 已提交
472 473
  /*pTmq->epStatus = 0;*/
  /*pTmq->epSkipCnt = 0;*/
L
Liu Jicong 已提交
474

L
Liu Jicong 已提交
475 476 477
  // set conf
  strcpy(pTmq->clientId, conf->clientId);
  strcpy(pTmq->groupId, conf->groupId);
L
Liu Jicong 已提交
478
  pTmq->autoCommit = conf->autoCommit;
L
Liu Jicong 已提交
479
  pTmq->autoCommitInterval = conf->autoCommitInterval;
480
  pTmq->commit_cb = conf->commitCb;
L
Liu Jicong 已提交
481 482
  pTmq->resetOffsetCfg = conf->resetOffset;

L
Liu Jicong 已提交
483
  // assign consumerId
L
Liu Jicong 已提交
484
  pTmq->consumerId = tGenIdPI64();
X
Xiaoyu Wang 已提交
485

L
Liu Jicong 已提交
486 487 488 489
  // init semaphore
  if (tsem_init(&pTmq->rspSem, 0, 0) != 0) {
    goto FAIL;
  }
L
Liu Jicong 已提交
490

L
Liu Jicong 已提交
491 492 493 494 495 496
  // init connection
  pTmq->pTscObj = taos_connect_internal(conf->ip, user, pass, NULL, NULL, conf->port, CONN_TYPE__TMQ);
  if (pTmq->pTscObj == NULL) {
    tsem_destroy(&pTmq->rspSem);
    goto FAIL;
  }
L
Liu Jicong 已提交
497

498
  return pTmq;
L
Liu Jicong 已提交
499 500 501 502 503 504 505 506

FAIL:
  if (pTmq->clientTopics) taosArrayDestroy(pTmq->clientTopics);
  if (pTmq->mqueue) taosCloseQueue(pTmq->mqueue);
  if (pTmq->delayedTask) taosCloseQueue(pTmq->delayedTask);
  if (pTmq->qall) taosFreeQall(pTmq->qall);
  taosMemoryFree(pTmq);
  return NULL;
507 508
}

L
Liu Jicong 已提交
509 510 511 512
tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, int32_t async) {
  // TODO: add read write lock
  SRequestObj*   pRequest = NULL;
  tmq_resp_err_t resp = TMQ_RESP_ERR__SUCCESS;
L
Liu Jicong 已提交
513 514
  // build msg
  // send to mnode
L
Liu Jicong 已提交
515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534
  SMqCMCommitOffsetReq req;
  SArray*              pArray = NULL;

  if (offsets == NULL) {
    pArray = taosArrayInit(0, sizeof(SMqOffset));
    for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
      SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
      for (int j = 0; j < taosArrayGetSize(pTopic->vgs); j++) {
        SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j);
        SMqOffset    offset;
        strcpy(offset.topicName, pTopic->topicName);
        strcpy(offset.cgroup, tmq->groupId);
        offset.vgId = pVg->vgId;
        offset.offset = pVg->currentOffset;
        taosArrayPush(pArray, &offset);
      }
    }
    req.num = pArray->size;
    req.offsets = pArray->pData;
  } else {
L
Liu Jicong 已提交
535 536
    req.num = taosArrayGetSize(&offsets->container);
    req.offsets = (SMqOffset*)offsets->container.pData;
L
Liu Jicong 已提交
537
  }
L
Liu Jicong 已提交
538 539 540 541

  SCoder encoder;

  tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER);
L
Liu Jicong 已提交
542
  tEncodeSMqCMCommitOffsetReq(&encoder, &req);
L
Liu Jicong 已提交
543
  int32_t tlen = encoder.pos;
wafwerar's avatar
wafwerar 已提交
544
  void*   buf = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
545 546 547 548 549 550 551
  if (buf == NULL) {
    tCoderClear(&encoder);
    return -1;
  }
  tCoderClear(&encoder);

  tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, tlen, TD_ENCODER);
L
Liu Jicong 已提交
552
  tEncodeSMqCMCommitOffsetReq(&encoder, &req);
L
Liu Jicong 已提交
553 554
  tCoderClear(&encoder);

L
Liu Jicong 已提交
555
  pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_MND_MQ_COMMIT_OFFSET);
L
Liu Jicong 已提交
556 557 558 559
  if (pRequest == NULL) {
    tscError("failed to malloc request");
  }

wafwerar's avatar
wafwerar 已提交
560
  SMqCommitCbParam* pParam = taosMemoryMalloc(sizeof(SMqCommitCbParam));
L
Liu Jicong 已提交
561 562 563 564 565
  if (pParam == NULL) {
    return -1;
  }
  pParam->tmq = tmq;
  tsem_init(&pParam->rspSem, 0, 0);
L
fix  
Liu Jicong 已提交
566
  pParam->async = async;
L
Liu Jicong 已提交
567

X
Xiaoyu Wang 已提交
568 569 570 571 572
  pRequest->body.requestMsg = (SDataBuf){
      .pData = buf,
      .len = tlen,
      .handle = NULL,
  };
L
Liu Jicong 已提交
573 574

  SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest);
L
Liu Jicong 已提交
575 576
  sendInfo->param = pParam;
  sendInfo->fp = tmqCommitCb;
L
Liu Jicong 已提交
577 578 579 580 581
  SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);

  int64_t transporterId = 0;
  asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);

L
Liu Jicong 已提交
582 583 584 585
  if (!async) {
    tsem_wait(&pParam->rspSem);
    resp = pParam->rspErr;
  }
L
Liu Jicong 已提交
586

L
fix  
Liu Jicong 已提交
587
  tsem_destroy(&pParam->rspSem);
wafwerar's avatar
wafwerar 已提交
588
  taosMemoryFree(pParam);
L
fix  
Liu Jicong 已提交
589

L
Liu Jicong 已提交
590 591 592 593 594
  if (pArray) {
    taosArrayDestroy(pArray);
  }

  return resp;
L
Liu Jicong 已提交
595 596
}

L
Liu Jicong 已提交
597 598 599 600 601 602
tmq_resp_err_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
  const SArray*   container = &topic_list->container;
  int32_t         sz = taosArrayGetSize(container);
  void*           buf = NULL;
  SCMSubscribeReq req = {0};
  int32_t         code = -1;
603 604

  req.consumerId = tmq->consumerId;
L
Liu Jicong 已提交
605
  tstrncpy(req.cgroup, tmq->groupId, TSDB_CGROUP_LEN);
606
  req.topicNames = taosArrayInit(sz, sizeof(void*));
L
Liu Jicong 已提交
607
  if (req.topicNames == NULL) goto FAIL;
608

L
Liu Jicong 已提交
609 610
  for (int32_t i = 0; i < sz; i++) {
    char* topic = taosArrayGetP(container, i);
611 612

    SName name = {0};
L
Liu Jicong 已提交
613
    tNameSetDbName(&name, tmq->pTscObj->acctId, topic, strlen(topic));
614

L
Liu Jicong 已提交
615 616 617
    char* topicFName = taosMemoryCalloc(1, TSDB_TOPIC_FNAME_LEN);
    if (topicFName == NULL) {
      goto FAIL;
618
    }
L
Liu Jicong 已提交
619
    tNameExtractFullName(&name, topicFName);
620

L
Liu Jicong 已提交
621 622 623
    tscDebug("subscribe topic: %s", topicFName);

    taosArrayPush(req.topicNames, &topicFName);
624 625
  }

L
Liu Jicong 已提交
626 627 628 629
  int32_t tlen = tSerializeSCMSubscribeReq(NULL, &req);
  buf = taosMemoryMalloc(tlen);
  if (buf == NULL) goto FAIL;

630 631 632
  void* abuf = buf;
  tSerializeSCMSubscribeReq(&abuf, &req);

L
Liu Jicong 已提交
633 634
  SMsgSendInfo* sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo));
  if (sendInfo == NULL) goto FAIL;
635

X
Xiaoyu Wang 已提交
636 637 638 639
  SMqSubscribeCbParam param = {
      .rspErr = TMQ_RESP_ERR__SUCCESS,
      .tmq = tmq,
  };
L
Liu Jicong 已提交
640

L
Liu Jicong 已提交
641 642 643
  if (tsem_init(&param.rspSem, 0, 0) != 0) goto FAIL;

  sendInfo->msgInfo = (SDataBuf){
X
Xiaoyu Wang 已提交
644 645 646 647
      .pData = buf,
      .len = tlen,
      .handle = NULL,
  };
648

L
Liu Jicong 已提交
649 650
  sendInfo->requestId = generateRequestId();
  sendInfo->requestObjRefId = 0;
L
Liu Jicong 已提交
651 652
  sendInfo->param = &param;
  sendInfo->fp = tmqSubscribeCb;
L
Liu Jicong 已提交
653 654
  sendInfo->msgType = TDMT_MND_SUBSCRIBE;

655 656 657 658 659
  SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);

  int64_t transporterId = 0;
  asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);

L
Liu Jicong 已提交
660 661 662
  // avoid double free if msg is sent
  buf = NULL;

L
Liu Jicong 已提交
663 664
  tsem_wait(&param.rspSem);
  tsem_destroy(&param.rspSem);
665

L
Liu Jicong 已提交
666 667 668 669
  code = param.rspErr;
  if (code != 0) goto FAIL;

  // TODO: add max retry cnt
L
Liu Jicong 已提交
670 671
  while (TSDB_CODE_MND_CONSUMER_NOT_READY == tmqAskEp(tmq, false)) {
    tscDebug("not ready, retry");
L
Liu Jicong 已提交
672 673
    taosMsleep(500);
  }
674

L
Liu Jicong 已提交
675 676 677 678 679 680 681 682
  // init hb timer
  tmq->hbTimer = taosTmrStart(tmqAssignDelayedHbTask, 1000, tmq, tmqMgmt.timer);

  // init auto commit timer
  if (tmq->autoCommit) {
    tmq->commitTimer = taosTmrStart(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, tmq, tmqMgmt.timer);
  }

L
Liu Jicong 已提交
683 684 685 686 687 688 689
  code = 0;
FAIL:
  if (req.topicNames != NULL) taosArrayDestroyP(req.topicNames, taosMemoryFree);
  if (code != 0) {
    taosMemoryFree(buf);
  }
  return code;
690 691
}

L
Liu Jicong 已提交
692 693
void tmq_conf_set_offset_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb) {
  //
694
  conf->commitCb = cb;
L
Liu Jicong 已提交
695
}
696

L
Liu Jicong 已提交
697 698 699 700 701 702 703 704 705 706 707 708 709 710 711
TAOS_RES* tmq_create_stream(TAOS* taos, const char* streamName, const char* tbName, const char* sql) {
  STscObj*     pTscObj = (STscObj*)taos;
  SRequestObj* pRequest = NULL;
  SQuery*      pQueryNode = NULL;
  char*        astStr = NULL;
  int32_t      sqlLen;

  terrno = TSDB_CODE_SUCCESS;
  if (taos == NULL || streamName == NULL || sql == NULL) {
    tscError("invalid parameters for creating stream, connObj:%p, stream name:%s, sql:%s", taos, streamName, sql);
    terrno = TSDB_CODE_TSC_INVALID_INPUT;
    goto _return;
  }
  sqlLen = strlen(sql);

L
Liu Jicong 已提交
712 713
  if (strlen(tbName) >= TSDB_TABLE_NAME_LEN) {
    tscError("output tb name too long, max length:%d", TSDB_TABLE_NAME_LEN - 1);
L
Liu Jicong 已提交
714 715 716 717 718 719 720 721 722 723 724 725
    terrno = TSDB_CODE_TSC_INVALID_INPUT;
    goto _return;
  }

  if (sqlLen > TSDB_MAX_ALLOWED_SQL_LEN) {
    tscError("sql string exceeds max length:%d", TSDB_MAX_ALLOWED_SQL_LEN);
    terrno = TSDB_CODE_TSC_EXCEED_SQL_LIMIT;
    goto _return;
  }

  tscDebug("start to create stream: %s", streamName);

H
Haojun Liao 已提交
726
  int32_t code = 0;
L
Liu Jicong 已提交
727
  CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
D
stmt  
dapan1121 已提交
728
  CHECK_CODE_GOTO(parseSql(pRequest, false, &pQueryNode, NULL), _return);
L
Liu Jicong 已提交
729 730 731 732 733 734 735 736 737 738 739 740 741 742
  CHECK_CODE_GOTO(nodesNodeToString(pQueryNode->pRoot, false, &astStr, NULL), _return);

  /*printf("%s\n", pStr);*/

  SName name = {.acctId = pTscObj->acctId, .type = TSDB_TABLE_NAME_T};
  strcpy(name.dbname, pRequest->pDb);
  strcpy(name.tname, streamName);

  SCMCreateStreamReq req = {
      .igExists = 1,
      .ast = astStr,
      .sql = (char*)sql,
  };
  tNameExtractFullName(&name, req.name);
L
Liu Jicong 已提交
743
  strcpy(req.outputSTbName, tbName);
L
Liu Jicong 已提交
744 745

  int   tlen = tSerializeSCMCreateStreamReq(NULL, 0, &req);
wafwerar's avatar
wafwerar 已提交
746
  void* buf = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768
  if (buf == NULL) {
    goto _return;
  }

  tSerializeSCMCreateStreamReq(buf, tlen, &req);

  pRequest->body.requestMsg = (SDataBuf){
      .pData = buf,
      .len = tlen,
      .handle = NULL,
  };
  pRequest->type = TDMT_MND_CREATE_STREAM;

  SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest);
  SEpSet        epSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

  int64_t transporterId = 0;
  asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);

  tsem_wait(&pRequest->body.rspSem);

_return:
wafwerar's avatar
wafwerar 已提交
769
  taosMemoryFreeClear(astStr);
L
Liu Jicong 已提交
770 771 772 773 774 775 776 777 778 779 780 781
  qDestroyQuery(pQueryNode);
  /*if (sendInfo != NULL) {*/
  /*destroySendMsgInfo(sendInfo);*/
  /*}*/

  if (pRequest != NULL && terrno != TSDB_CODE_SUCCESS) {
    pRequest->code = terrno;
  }

  return pRequest;
}

L
Liu Jicong 已提交
782
#if 0
L
Liu Jicong 已提交
783 784
int32_t tmqGetSkipLogNum(tmq_message_t* tmq_message) {
  if (tmq_message == NULL) return 0;
L
Liu Jicong 已提交
785
  SMqPollRsp* pRsp = &tmq_message->msg;
L
Liu Jicong 已提交
786 787
  return pRsp->skipLogNum;
}
L
Liu Jicong 已提交
788
#endif
L
Liu Jicong 已提交
789 790

int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
X
Xiaoyu Wang 已提交
791 792
  SMqPollCbParam* pParam = (SMqPollCbParam*)param;
  SMqClientVg*    pVg = pParam->pVg;
L
Liu Jicong 已提交
793
  SMqClientTopic* pTopic = pParam->pTopic;
X
Xiaoyu Wang 已提交
794
  tmq_t*          tmq = pParam->tmq;
L
Liu Jicong 已提交
795
  if (code != 0) {
L
Liu Jicong 已提交
796
    tscWarn("msg discard from vg %d, epoch %d, code:%x", pParam->vgId, pParam->epoch, code);
L
fix txn  
Liu Jicong 已提交
797
    goto CREATE_MSG_FAIL;
L
Liu Jicong 已提交
798 799
  }

X
Xiaoyu Wang 已提交
800 801 802
  int32_t msgEpoch = ((SMqRspHead*)pMsg->pData)->epoch;
  int32_t tmqEpoch = atomic_load_32(&tmq->epoch);
  if (msgEpoch < tmqEpoch) {
L
Liu Jicong 已提交
803
    // do not write into queue since updating epoch reset
L
Liu Jicong 已提交
804 805
    tscWarn("msg discard from vg %d since from earlier epoch, rsp epoch %d, current epoch %d", pParam->vgId, msgEpoch,
            tmqEpoch);
806
    tsem_post(&tmq->rspSem);
X
Xiaoyu Wang 已提交
807 808 809 810
    return 0;
  }

  if (msgEpoch != tmqEpoch) {
L
Liu Jicong 已提交
811
    tscWarn("mismatch rsp from vg %d, epoch %d, current epoch %d", pParam->vgId, msgEpoch, tmqEpoch);
X
Xiaoyu Wang 已提交
812 813
  }

L
Liu Jicong 已提交
814 815
  SMqPollRspWrapper* pRspWrapper = taosAllocateQitem(sizeof(SMqPollRspWrapper));
  if (pRspWrapper == NULL) {
L
Liu Jicong 已提交
816
    tscWarn("msg discard from vg %d, epoch %d since out of memory", pParam->vgId, pParam->epoch);
L
fix txn  
Liu Jicong 已提交
817
    goto CREATE_MSG_FAIL;
L
Liu Jicong 已提交
818
  }
L
Liu Jicong 已提交
819

L
Liu Jicong 已提交
820 821 822
  pRspWrapper->tmqRspType = TMQ_MSG_TYPE__POLL_RSP;
  pRspWrapper->vgHandle = pVg;
  pRspWrapper->topicHandle = pTopic;
L
Liu Jicong 已提交
823

L
Liu Jicong 已提交
824 825
  memcpy(&pRspWrapper->msg, pMsg->pData, sizeof(SMqRspHead));

L
Liu Jicong 已提交
826
  tDecodeSMqDataBlkRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pRspWrapper->msg);
L
Liu Jicong 已提交
827

L
Liu Jicong 已提交
828 829
  tscDebug("consumer %ld recv poll: vg %d, req offset %ld, rsp offset %ld", tmq->consumerId, pVg->vgId,
           pRspWrapper->msg.reqOffset, pRspWrapper->msg.rspOffset);
L
fix  
Liu Jicong 已提交
830

L
Liu Jicong 已提交
831
  taosWriteQitem(tmq->mqueue, pRspWrapper);
832
  tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
833

L
Liu Jicong 已提交
834
  return 0;
L
fix txn  
Liu Jicong 已提交
835
CREATE_MSG_FAIL:
L
Liu Jicong 已提交
836 837 838
  if (pParam->epoch == tmq->epoch) {
    atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
  }
839
  tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
840
  return -1;
841 842
}

L
Liu Jicong 已提交
843
bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) {
L
Liu Jicong 已提交
844
  /*printf("call update ep %d\n", epoch);*/
X
Xiaoyu Wang 已提交
845
  bool    set = false;
L
Liu Jicong 已提交
846 847
  int32_t topicNumGet = taosArrayGetSize(pRsp->topics);
  char    vgKey[TSDB_TOPIC_FNAME_LEN + 22];
L
Liu Jicong 已提交
848 849
  tscDebug("consumer %ld update ep epoch %d to epoch %d, topic num: %d", tmq->consumerId, tmq->epoch, epoch,
           topicNumGet);
L
Liu Jicong 已提交
850 851 852 853 854 855 856 857 858 859 860 861
  SArray* newTopics = taosArrayInit(topicNumGet, sizeof(SMqClientTopic));
  if (newTopics == NULL) {
    return false;
  }
  SHashObj* pHash = taosHashInit(64, MurmurHash3_32, false, HASH_NO_LOCK);
  if (pHash == NULL) {
    taosArrayDestroy(newTopics);
    return false;
  }

  // find topic, build hash
  for (int32_t i = 0; i < topicNumGet; i++) {
X
Xiaoyu Wang 已提交
862 863
    SMqClientTopic topic = {0};
    SMqSubTopicEp* pTopicEp = taosArrayGet(pRsp->topics, i);
L
Liu Jicong 已提交
864
    topic.schema = pTopicEp->schema;
L
Liu Jicong 已提交
865
    taosHashClear(pHash);
X
Xiaoyu Wang 已提交
866
    topic.topicName = strdup(pTopicEp->topic);
L
Liu Jicong 已提交
867

L
Liu Jicong 已提交
868
    tscDebug("consumer %ld update topic: %s", tmq->consumerId, topic.topicName);
L
Liu Jicong 已提交
869 870 871 872 873 874
    int32_t topicNumCur = taosArrayGetSize(tmq->clientTopics);
    for (int32_t j = 0; j < topicNumCur; j++) {
      // find old topic
      SMqClientTopic* pTopicCur = taosArrayGet(tmq->clientTopics, j);
      if (pTopicCur->vgs && strcmp(pTopicCur->topicName, pTopicEp->topic) == 0) {
        int32_t vgNumCur = taosArrayGetSize(pTopicCur->vgs);
L
Liu Jicong 已提交
875
        tscDebug("consumer %ld new vg num: %d", tmq->consumerId, vgNumCur);
L
Liu Jicong 已提交
876 877 878 879
        if (vgNumCur == 0) break;
        for (int32_t k = 0; k < vgNumCur; k++) {
          SMqClientVg* pVgCur = taosArrayGet(pTopicCur->vgs, k);
          sprintf(vgKey, "%s:%d", topic.topicName, pVgCur->vgId);
L
Liu Jicong 已提交
880
          tscDebug("consumer %ld epoch %d vg %d build %s", tmq->consumerId, epoch, pVgCur->vgId, vgKey);
L
Liu Jicong 已提交
881 882 883 884 885 886 887 888 889
          taosHashPut(pHash, vgKey, strlen(vgKey), &pVgCur->currentOffset, sizeof(int64_t));
        }
        break;
      }
    }

    int32_t vgNumGet = taosArrayGetSize(pTopicEp->vgs);
    topic.vgs = taosArrayInit(vgNumGet, sizeof(SMqClientVg));
    for (int32_t j = 0; j < vgNumGet; j++) {
X
Xiaoyu Wang 已提交
890
      SMqSubVgEp* pVgEp = taosArrayGet(pTopicEp->vgs, j);
L
Liu Jicong 已提交
891 892 893
      sprintf(vgKey, "%s:%d", topic.topicName, pVgEp->vgId);
      int64_t* pOffset = taosHashGet(pHash, vgKey, strlen(vgKey));
      int64_t  offset = pVgEp->offset;
L
Liu Jicong 已提交
894
      tscDebug("consumer %ld epoch %d vg %d offset og to %ld", tmq->consumerId, epoch, pVgEp->vgId, offset);
L
Liu Jicong 已提交
895 896
      if (pOffset != NULL) {
        offset = *pOffset;
L
Liu Jicong 已提交
897
        tscDebug("consumer %ld epoch %d vg %d found %s", tmq->consumerId, epoch, pVgEp->vgId, vgKey);
L
Liu Jicong 已提交
898
      }
L
Liu Jicong 已提交
899
      tscDebug("consumer %ld epoch %d vg %d offset set to %ld", tmq->consumerId, epoch, pVgEp->vgId, offset);
X
Xiaoyu Wang 已提交
900 901
      SMqClientVg clientVg = {
          .pollCnt = 0,
L
Liu Jicong 已提交
902
          .currentOffset = offset,
X
Xiaoyu Wang 已提交
903 904 905
          .vgId = pVgEp->vgId,
          .epSet = pVgEp->epSet,
          .vgStatus = TMQ_VG_STATUS__IDLE,
L
Liu Jicong 已提交
906
          .vgSkipCnt = 0,
X
Xiaoyu Wang 已提交
907 908 909 910
      };
      taosArrayPush(topic.vgs, &clientVg);
      set = true;
    }
L
Liu Jicong 已提交
911
    taosArrayPush(newTopics, &topic);
X
Xiaoyu Wang 已提交
912
  }
L
Liu Jicong 已提交
913
  if (tmq->clientTopics) taosArrayDestroy(tmq->clientTopics);
L
Liu Jicong 已提交
914
  taosHashCleanup(pHash);
L
Liu Jicong 已提交
915
  tmq->clientTopics = newTopics;
916 917 918 919 920 921

  if (taosArrayGetSize(tmq->clientTopics) == 0)
    atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__NO_TOPIC);
  else
    atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY);

X
Xiaoyu Wang 已提交
922 923 924 925
  atomic_store_32(&tmq->epoch, epoch);
  return set;
}

L
Liu Jicong 已提交
926
int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
927
  SMqAskEpCbParam* pParam = (SMqAskEpCbParam*)param;
L
Liu Jicong 已提交
928
  tmq_t*           tmq = pParam->tmq;
L
Liu Jicong 已提交
929
  pParam->code = code;
930
  if (code != 0) {
L
Liu Jicong 已提交
931
    tscError("consumer %ld get topic endpoint error, not ready, wait:%d", tmq->consumerId, pParam->async);
L
Liu Jicong 已提交
932
    goto END;
933
  }
L
Liu Jicong 已提交
934

L
Liu Jicong 已提交
935
  // tmq's epoch is monotonically increase,
L
Liu Jicong 已提交
936
  // so it's safe to discard any old epoch msg.
L
Liu Jicong 已提交
937
  // Epoch will only increase when received newer epoch ep msg
L
Liu Jicong 已提交
938 939
  SMqRspHead* head = pMsg->pData;
  int32_t     epoch = atomic_load_32(&tmq->epoch);
L
temp  
Liu Jicong 已提交
940
  tscDebug("consumer %ld recv ep, msg epoch %d, current epoch %d", tmq->consumerId, head->epoch, epoch);
L
Liu Jicong 已提交
941 942
  if (head->epoch <= epoch) {
    goto END;
943
  }
L
Liu Jicong 已提交
944

L
Liu Jicong 已提交
945 946 947
  if (!pParam->async) {
    SMqAskEpRsp rsp;
    tDecodeSMqAskEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &rsp);
X
Xiaoyu Wang 已提交
948 949
    /*printf("rsp epoch %ld sz %ld\n", rsp.epoch, rsp.topics->size);*/
    /*printf("tmq epoch %ld sz %ld\n", tmq->epoch, tmq->clientTopics->size);*/
950
    tmqUpdateEp(tmq, head->epoch, &rsp);
L
Liu Jicong 已提交
951
    tDeleteSMqAskEpRsp(&rsp);
X
Xiaoyu Wang 已提交
952
  } else {
L
Liu Jicong 已提交
953 954
    SMqAskEpRspWrapper* pWrapper = taosAllocateQitem(sizeof(SMqAskEpRspWrapper));
    if (pWrapper == NULL) {
X
Xiaoyu Wang 已提交
955
      terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
956 957
      code = -1;
      goto END;
X
Xiaoyu Wang 已提交
958
    }
L
Liu Jicong 已提交
959 960 961
    pWrapper->tmqRspType = TMQ_MSG_TYPE__EP_RSP;
    pWrapper->epoch = head->epoch;
    memcpy(&pWrapper->msg, pMsg->pData, sizeof(SMqRspHead));
L
Liu Jicong 已提交
962
    tDecodeSMqAskEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pWrapper->msg);
L
Liu Jicong 已提交
963

L
Liu Jicong 已提交
964
    taosWriteQitem(tmq->mqueue, pWrapper);
965
    tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
966
    taosMemoryFree(pParam);
967
  }
L
Liu Jicong 已提交
968 969

END:
L
Liu Jicong 已提交
970
  /*atomic_store_8(&tmq->epStatus, 0);*/
L
Liu Jicong 已提交
971
  if (!pParam->async) {
L
Liu Jicong 已提交
972 973 974
    tsem_post(&pParam->rspSem);
  }
  return code;
975 976
}

L
Liu Jicong 已提交
977
int32_t tmqAskEp(tmq_t* tmq, bool async) {
L
Liu Jicong 已提交
978
  int32_t code = 0;
L
Liu Jicong 已提交
979
#if 0
L
Liu Jicong 已提交
980
  int8_t  epStatus = atomic_val_compare_exchange_8(&tmq->epStatus, 0, 1);
L
fix txn  
Liu Jicong 已提交
981
  if (epStatus == 1) {
L
temp  
Liu Jicong 已提交
982
    int32_t epSkipCnt = atomic_add_fetch_32(&tmq->epSkipCnt, 1);
L
Liu Jicong 已提交
983
    tscTrace("consumer %ld skip ask ep cnt %d", tmq->consumerId, epSkipCnt);
L
Liu Jicong 已提交
984
    if (epSkipCnt < 5000) return 0;
L
fix txn  
Liu Jicong 已提交
985
  }
L
temp  
Liu Jicong 已提交
986
  atomic_store_32(&tmq->epSkipCnt, 0);
L
Liu Jicong 已提交
987
#endif
L
Liu Jicong 已提交
988 989
  int32_t      tlen = sizeof(SMqAskEpReq);
  SMqAskEpReq* req = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
990
  if (req == NULL) {
L
Liu Jicong 已提交
991
    tscError("failed to malloc get subscribe ep buf");
L
Liu Jicong 已提交
992
    /*atomic_store_8(&tmq->epStatus, 0);*/
L
Liu Jicong 已提交
993
    return -1;
L
Liu Jicong 已提交
994
  }
L
Liu Jicong 已提交
995 996 997
  req->consumerId = htobe64(tmq->consumerId);
  req->epoch = htonl(tmq->epoch);
  strcpy(req->cgroup, tmq->groupId);
998

wafwerar's avatar
wafwerar 已提交
999
  SMqAskEpCbParam* pParam = taosMemoryMalloc(sizeof(SMqAskEpCbParam));
L
Liu Jicong 已提交
1000 1001
  if (pParam == NULL) {
    tscError("failed to malloc subscribe param");
wafwerar's avatar
wafwerar 已提交
1002
    taosMemoryFree(req);
L
Liu Jicong 已提交
1003
    /*atomic_store_8(&tmq->epStatus, 0);*/
L
Liu Jicong 已提交
1004
    return -1;
L
Liu Jicong 已提交
1005 1006
  }
  pParam->tmq = tmq;
L
Liu Jicong 已提交
1007
  pParam->async = async;
X
Xiaoyu Wang 已提交
1008
  tsem_init(&pParam->rspSem, 0, 0);
1009

wafwerar's avatar
wafwerar 已提交
1010
  SMsgSendInfo* sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo));
L
Liu Jicong 已提交
1011 1012
  if (sendInfo == NULL) {
    tsem_destroy(&pParam->rspSem);
wafwerar's avatar
wafwerar 已提交
1013 1014
    taosMemoryFree(pParam);
    taosMemoryFree(req);
L
Liu Jicong 已提交
1015
    /*atomic_store_8(&tmq->epStatus, 0);*/
L
Liu Jicong 已提交
1016 1017 1018 1019 1020 1021 1022 1023 1024 1025
    return -1;
  }

  sendInfo->msgInfo = (SDataBuf){
      .pData = req,
      .len = tlen,
      .handle = NULL,
  };

  sendInfo->requestId = generateRequestId();
L
Liu Jicong 已提交
1026 1027 1028
  sendInfo->requestObjRefId = 0;
  sendInfo->param = pParam;
  sendInfo->fp = tmqAskEpCb;
L
Liu Jicong 已提交
1029
  sendInfo->msgType = TDMT_MND_MQ_ASK_EP;
1030

L
Liu Jicong 已提交
1031 1032
  SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);

L
add log  
Liu Jicong 已提交
1033 1034
  tscDebug("consumer %ld ask ep", tmq->consumerId);

L
Liu Jicong 已提交
1035 1036
  int64_t transporterId = 0;
  asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);
1037

L
Liu Jicong 已提交
1038
  if (!async) {
L
Liu Jicong 已提交
1039 1040 1041 1042 1043
    tsem_wait(&pParam->rspSem);
    code = pParam->code;
    taosMemoryFree(pParam);
  }
  return code;
1044 1045
}

L
Liu Jicong 已提交
1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059
tmq_resp_err_t tmq_seek(tmq_t* tmq, const tmq_topic_vgroup_t* offset) {
  const SMqOffset* pOffset = &offset->offset;
  if (strcmp(pOffset->cgroup, tmq->groupId) != 0) {
    return TMQ_RESP_ERR__FAIL;
  }
  int32_t sz = taosArrayGetSize(tmq->clientTopics);
  for (int32_t i = 0; i < sz; i++) {
    SMqClientTopic* clientTopic = taosArrayGet(tmq->clientTopics, i);
    if (strcmp(clientTopic->topicName, pOffset->topicName) == 0) {
      int32_t vgSz = taosArrayGetSize(clientTopic->vgs);
      for (int32_t j = 0; j < vgSz; j++) {
        SMqClientVg* pVg = taosArrayGet(clientTopic->vgs, j);
        if (pVg->vgId == pOffset->vgId) {
          pVg->currentOffset = pOffset->offset;
L
Liu Jicong 已提交
1060
          tmqClearUnhandleMsg(tmq);
L
Liu Jicong 已提交
1061 1062 1063 1064 1065 1066 1067 1068
          return TMQ_RESP_ERR__SUCCESS;
        }
      }
    }
  }
  return TMQ_RESP_ERR__FAIL;
}

L
Liu Jicong 已提交
1069
SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t waitTime, SMqClientTopic* pTopic, SMqClientVg* pVg) {
L
Liu Jicong 已提交
1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080
  int64_t reqOffset;
  if (pVg->currentOffset >= 0) {
    reqOffset = pVg->currentOffset;
  } else {
    if (tmq->resetOffsetCfg == TMQ_CONF__RESET_OFFSET__NONE) {
      tscError("unable to poll since no committed offset but reset offset is set to none");
      return NULL;
    }
    reqOffset = tmq->resetOffsetCfg;
  }

L
Liu Jicong 已提交
1081
  SMqPollReq* pReq = taosMemoryMalloc(sizeof(SMqPollReq));
1082 1083 1084
  if (pReq == NULL) {
    return NULL;
  }
L
Liu Jicong 已提交
1085

L
Liu Jicong 已提交
1086 1087 1088 1089 1090 1091 1092
  /*strcpy(pReq->topic, pTopic->topicName);*/
  /*strcpy(pReq->cgroup, tmq->groupId);*/

  int32_t tlen = strlen(tmq->groupId);
  memcpy(pReq->subKey, tmq->groupId, tlen);
  pReq->subKey[tlen] = TMQ_SEPARATOR;
  strcpy(pReq->subKey + tlen + 1, pTopic->topicName);
1093

1094
  pReq->waitTime = waitTime;
L
Liu Jicong 已提交
1095
  pReq->consumerId = tmq->consumerId;
X
Xiaoyu Wang 已提交
1096
  pReq->epoch = tmq->epoch;
L
Liu Jicong 已提交
1097
  pReq->currentOffset = reqOffset;
L
Liu Jicong 已提交
1098
  pReq->reqId = generateRequestId();
1099 1100

  pReq->head.vgId = htonl(pVg->vgId);
L
Liu Jicong 已提交
1101
  pReq->head.contLen = htonl(sizeof(SMqPollReq));
1102 1103 1104
  return pReq;
}

L
Liu Jicong 已提交
1105 1106 1107
SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper) {
  SMqRspObj* pRspObj = taosMemoryCalloc(1, sizeof(SMqRspObj));
  pRspObj->resType = RES_TYPE__TMQ;
L
Liu Jicong 已提交
1108
  strncpy(pRspObj->topic, pWrapper->topicHandle->topicName, TSDB_TOPIC_FNAME_LEN);
L
Liu Jicong 已提交
1109
  pRspObj->vgId = pWrapper->vgHandle->vgId;
L
Liu Jicong 已提交
1110 1111
  pRspObj->resIter = -1;
  memcpy(&pRspObj->rsp, &pWrapper->msg, sizeof(SMqDataBlkRsp));
L
Liu Jicong 已提交
1112

L
Liu Jicong 已提交
1113 1114
  pRspObj->resInfo.totalRows = 0;
  pRspObj->resInfo.precision = TSDB_TIME_PRECISION_MILLI;
L
Liu Jicong 已提交
1115 1116 1117
  if (!pWrapper->msg.withSchema) {
    setResSchemaInfo(&pRspObj->resInfo, pWrapper->topicHandle->schema.pSchema, pWrapper->topicHandle->schema.nCols);
  }
L
Liu Jicong 已提交
1118

L
Liu Jicong 已提交
1119 1120
  taosFreeQitem(pWrapper);
  return pRspObj;
X
Xiaoyu Wang 已提交
1121 1122
}

1123
int32_t tmqPollImpl(tmq_t* tmq, int64_t waitTime) {
L
fix  
Liu Jicong 已提交
1124
  /*printf("call poll\n");*/
X
Xiaoyu Wang 已提交
1125 1126 1127 1128 1129 1130
  for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
    SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
    for (int j = 0; j < taosArrayGetSize(pTopic->vgs); j++) {
      SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j);
      int32_t      vgStatus = atomic_val_compare_exchange_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE, TMQ_VG_STATUS__WAIT);
      if (vgStatus != TMQ_VG_STATUS__IDLE) {
L
Liu Jicong 已提交
1131
        int32_t vgSkipCnt = atomic_add_fetch_32(&pVg->vgSkipCnt, 1);
L
Liu Jicong 已提交
1132
        tscTrace("consumer %ld epoch %d skip vg %d skip cnt %d", tmq->consumerId, tmq->epoch, pVg->vgId, vgSkipCnt);
X
Xiaoyu Wang 已提交
1133
        continue;
L
Liu Jicong 已提交
1134
        /*if (vgSkipCnt < 10000) continue;*/
L
temp  
Liu Jicong 已提交
1135 1136 1137 1138 1139 1140 1141
#if 0
        if (skipCnt < 30000) {
          continue;
        } else {
        tscDebug("consumer %ld skip vg %d skip too much reset", tmq->consumerId, pVg->vgId);
        }
#endif
X
Xiaoyu Wang 已提交
1142
      }
L
Liu Jicong 已提交
1143
      atomic_store_32(&pVg->vgSkipCnt, 0);
L
Liu Jicong 已提交
1144
      SMqPollReq* pReq = tmqBuildConsumeReqImpl(tmq, waitTime, pTopic, pVg);
X
Xiaoyu Wang 已提交
1145 1146
      if (pReq == NULL) {
        atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
1147
        tsem_post(&tmq->rspSem);
X
Xiaoyu Wang 已提交
1148 1149
        return -1;
      }
wafwerar's avatar
wafwerar 已提交
1150
      SMqPollCbParam* pParam = taosMemoryMalloc(sizeof(SMqPollCbParam));
L
Liu Jicong 已提交
1151
      if (pParam == NULL) {
wafwerar's avatar
wafwerar 已提交
1152
        taosMemoryFree(pReq);
X
Xiaoyu Wang 已提交
1153
        atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
1154
        tsem_post(&tmq->rspSem);
X
Xiaoyu Wang 已提交
1155 1156
        return -1;
      }
L
Liu Jicong 已提交
1157 1158
      pParam->tmq = tmq;
      pParam->pVg = pVg;
L
Liu Jicong 已提交
1159
      pParam->pTopic = pTopic;
L
Liu Jicong 已提交
1160
      pParam->vgId = pVg->vgId;
L
Liu Jicong 已提交
1161 1162
      pParam->epoch = tmq->epoch;

wafwerar's avatar
wafwerar 已提交
1163
      SMsgSendInfo* sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo));
L
Liu Jicong 已提交
1164
      if (sendInfo == NULL) {
wafwerar's avatar
wafwerar 已提交
1165 1166
        taosMemoryFree(pReq);
        taosMemoryFree(pParam);
L
Liu Jicong 已提交
1167
        atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
1168
        tsem_post(&tmq->rspSem);
L
Liu Jicong 已提交
1169 1170 1171 1172
        return -1;
      }

      sendInfo->msgInfo = (SDataBuf){
X
Xiaoyu Wang 已提交
1173
          .pData = pReq,
L
Liu Jicong 已提交
1174
          .len = sizeof(SMqPollReq),
X
Xiaoyu Wang 已提交
1175 1176
          .handle = NULL,
      };
L
Liu Jicong 已提交
1177
      sendInfo->requestId = pReq->reqId;
X
Xiaoyu Wang 已提交
1178
      sendInfo->requestObjRefId = 0;
L
Liu Jicong 已提交
1179
      sendInfo->param = pParam;
X
Xiaoyu Wang 已提交
1180
      sendInfo->fp = tmqPollCb;
L
Liu Jicong 已提交
1181
      sendInfo->msgType = TDMT_VND_CONSUME;
X
Xiaoyu Wang 已提交
1182 1183

      int64_t transporterId = 0;
L
fix  
Liu Jicong 已提交
1184
      /*printf("send poll\n");*/
L
Liu Jicong 已提交
1185 1186
      tscDebug("consumer %ld send poll to %s : vg %d, epoch %d, req offset %ld, reqId %lu", tmq->consumerId,
               pTopic->topicName, pVg->vgId, tmq->epoch, pVg->currentOffset, pReq->reqId);
L
fix  
Liu Jicong 已提交
1187
      /*printf("send vg %d %ld\n", pVg->vgId, pVg->currentOffset);*/
X
Xiaoyu Wang 已提交
1188 1189 1190 1191 1192 1193 1194 1195
      asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, sendInfo);
      pVg->pollCnt++;
      tmq->pollCnt++;
    }
  }
  return 0;
}

L
Liu Jicong 已提交
1196 1197
int32_t tmqHandleNoPollRsp(tmq_t* tmq, SMqRspWrapper* rspWrapper, bool* pReset) {
  if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__EP_RSP) {
L
fix  
Liu Jicong 已提交
1198
    /*printf("ep %d %d\n", rspMsg->head.epoch, tmq->epoch);*/
L
Liu Jicong 已提交
1199 1200
    if (rspWrapper->epoch > atomic_load_32(&tmq->epoch)) {
      SMqAskEpRspWrapper* pEpRspWrapper = (SMqAskEpRspWrapper*)rspWrapper;
L
Liu Jicong 已提交
1201
      SMqAskEpRsp*        rspMsg = &pEpRspWrapper->msg;
L
Liu Jicong 已提交
1202
      tmqUpdateEp(tmq, rspWrapper->epoch, rspMsg);
L
temp  
Liu Jicong 已提交
1203
      /*tmqClearUnhandleMsg(tmq);*/
X
Xiaoyu Wang 已提交
1204 1205 1206 1207 1208 1209 1210 1211 1212 1213
      *pReset = true;
    } else {
      *pReset = false;
    }
  } else {
    return -1;
  }
  return 0;
}

1214
SMqRspObj* tmqHandleAllRsp(tmq_t* tmq, int64_t waitTime, bool pollIfReset) {
X
Xiaoyu Wang 已提交
1215
  while (1) {
L
Liu Jicong 已提交
1216 1217 1218
    SMqRspWrapper* rspWrapper = NULL;
    taosGetQitem(tmq->qall, (void**)&rspWrapper);
    if (rspWrapper == NULL) {
L
Liu Jicong 已提交
1219
      taosReadAllQitems(tmq->mqueue, tmq->qall);
L
Liu Jicong 已提交
1220 1221
      taosGetQitem(tmq->qall, (void**)&rspWrapper);
      if (rspWrapper == NULL) return NULL;
X
Xiaoyu Wang 已提交
1222 1223
    }

L
Liu Jicong 已提交
1224 1225
    if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_RSP) {
      SMqPollRspWrapper* pollRspWrapper = (SMqPollRspWrapper*)rspWrapper;
L
Liu Jicong 已提交
1226
      /*atomic_sub_fetch_32(&tmq->readyRequest, 1);*/
L
fix  
Liu Jicong 已提交
1227
      /*printf("handle poll rsp %d\n", rspMsg->head.mqMsgType);*/
L
Liu Jicong 已提交
1228
      if (pollRspWrapper->msg.head.epoch == atomic_load_32(&tmq->epoch)) {
L
fix  
Liu Jicong 已提交
1229
        /*printf("epoch match\n");*/
L
Liu Jicong 已提交
1230
        SMqClientVg* pVg = pollRspWrapper->vgHandle;
L
Liu Jicong 已提交
1231
        /*printf("vg %d offset %ld up to %ld\n", pVg->vgId, pVg->currentOffset, rspMsg->msg.rspOffset);*/
L
Liu Jicong 已提交
1232
        pVg->currentOffset = pollRspWrapper->msg.rspOffset;
X
Xiaoyu Wang 已提交
1233
        atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
L
Liu Jicong 已提交
1234
        if (pollRspWrapper->msg.blockNum == 0) {
L
Liu Jicong 已提交
1235 1236
          taosFreeQitem(pollRspWrapper);
          rspWrapper = NULL;
L
temp  
Liu Jicong 已提交
1237 1238
          continue;
        }
L
Liu Jicong 已提交
1239
        // build rsp
L
Liu Jicong 已提交
1240 1241
        SMqRspObj* pRsp = tmqBuildRspFromWrapper(pollRspWrapper);
        return pRsp;
X
Xiaoyu Wang 已提交
1242
      } else {
L
Liu Jicong 已提交
1243
        /*printf("epoch mismatch\n");*/
L
Liu Jicong 已提交
1244
        taosFreeQitem(pollRspWrapper);
X
Xiaoyu Wang 已提交
1245 1246
      }
    } else {
L
fix  
Liu Jicong 已提交
1247
      /*printf("handle ep rsp %d\n", rspMsg->head.mqMsgType);*/
X
Xiaoyu Wang 已提交
1248
      bool reset = false;
L
Liu Jicong 已提交
1249 1250
      tmqHandleNoPollRsp(tmq, rspWrapper, &reset);
      taosFreeQitem(rspWrapper);
X
Xiaoyu Wang 已提交
1251
      if (pollIfReset && reset) {
L
add log  
Liu Jicong 已提交
1252
        tscDebug("consumer %ld reset and repoll", tmq->consumerId);
1253
        tmqPollImpl(tmq, waitTime);
X
Xiaoyu Wang 已提交
1254 1255 1256 1257 1258
      }
    }
  }
}

1259
TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
L
Liu Jicong 已提交
1260 1261
  SMqRspObj* rspObj;
  int64_t    startTime = taosGetTimestampMs();
L
Liu Jicong 已提交
1262

1263
  rspObj = tmqHandleAllRsp(tmq, wait_time, false);
L
Liu Jicong 已提交
1264 1265
  if (rspObj) {
    return (TAOS_RES*)rspObj;
L
fix  
Liu Jicong 已提交
1266
  }
X
Xiaoyu Wang 已提交
1267

L
Liu Jicong 已提交
1268
  if (atomic_load_8(&tmq->status) == TMQ_CONSUMER_STATUS__INIT) {
1269 1270 1271
    return NULL;
  }

X
Xiaoyu Wang 已提交
1272
  while (1) {
L
Liu Jicong 已提交
1273
    tmqHandleAllDelayedTask(tmq);
1274
    tmqPollImpl(tmq, wait_time);
L
Liu Jicong 已提交
1275

1276
    rspObj = tmqHandleAllRsp(tmq, wait_time, false);
L
Liu Jicong 已提交
1277 1278
    if (rspObj) {
      return (TAOS_RES*)rspObj;
X
Xiaoyu Wang 已提交
1279
    }
1280
    if (wait_time != 0) {
X
Xiaoyu Wang 已提交
1281
      int64_t endTime = taosGetTimestampMs();
1282 1283
      int64_t leftTime = endTime - startTime;
      if (leftTime > wait_time) {
L
Liu Jicong 已提交
1284
        tscDebug("consumer %ld (epoch %d) timeout, no rsp", tmq->consumerId, tmq->epoch);
X
Xiaoyu Wang 已提交
1285 1286
        return NULL;
      }
1287
      tsem_timewait(&tmq->rspSem, leftTime * 1000);
X
Xiaoyu Wang 已提交
1288 1289 1290 1291
    }
  }
}

L
Liu Jicong 已提交
1292 1293 1294
tmq_resp_err_t tmq_consumer_close(tmq_t* tmq) {
  // TODO
  return TMQ_RESP_ERR__SUCCESS;
1295
}
L
Liu Jicong 已提交
1296 1297 1298 1299 1300 1301 1302

const char* tmq_err2str(tmq_resp_err_t err) {
  if (err == TMQ_RESP_ERR__SUCCESS) {
    return "success";
  }
  return "fail";
}
L
Liu Jicong 已提交
1303

L
Liu Jicong 已提交
1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320
char* tmq_get_topic_name(TAOS_RES* res) {
  if (TD_RES_TMQ(res)) {
    SMqRspObj* pRspObj = (SMqRspObj*)res;
    return pRspObj->topic;
  } else {
    return NULL;
  }
}

int32_t tmq_get_vgroup_id(TAOS_RES* res) {
  if (TD_RES_TMQ(res)) {
    SMqRspObj* pRspObj = (SMqRspObj*)res;
    return pRspObj->vgId;
  } else {
    return -1;
  }
}