xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix:tmq error if consume callback is earlier than consume

This commit is contained in:
wangmm0220 2023-04-23 17:36:59 +08:00
parent 6fc364db19
commit af65f9703b
1 changed files with 4 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
utils/test/c/tmqSim.c
View File

@ -690,11 +690,12 @@ int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) {
} }
static int32_t g_once_commit_flag = 0; static int32_t g_once_commit_flag = 0;
static int32_t g_once_consume_flag = 0;
static void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) { static void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) {
taosFprintfFile(g_fp, "tmq_commit_cb_print() commit %d\n", code); taosFprintfFile(g_fp, "tmq_commit_cb_print() commit %d\n", code);
if (0 == g_once_commit_flag) { if (g_once_consume_flag == 1 && 0 == g_once_commit_flag) {
g_once_commit_flag = 1; g_once_commit_flag = 1;
notifyMainScript((SThreadInfo*)param, (int32_t)NOTIFY_CMD_START_COMMIT); notifyMainScript((SThreadInfo*)param, (int32_t)NOTIFY_CMD_START_COMMIT);
} }
@ -773,8 +774,6 @@ int32_t saveConsumeResult(SThreadInfo* pInfo) {
void loop_consume(SThreadInfo* pInfo) { void loop_consume(SThreadInfo* pInfo) {
int32_t code; int32_t code;
int32_t once_flag = 0;
int64_t totalMsgs = 0; int64_t totalMsgs = 0;
int64_t totalRows = 0; int64_t totalRows = 0;
@ -834,8 +833,8 @@ void loop_consume(SThreadInfo* pInfo) {
lastTotalMsgs = totalMsgs; lastTotalMsgs = totalMsgs;
} }
if (0 == once_flag) { if (0 == g_once_consume_flag) {
once_flag = 1; g_once_consume_flag = 1;
notifyMainScript(pInfo, NOTIFY_CMD_START_CONSUM); notifyMainScript(pInfo, NOTIFY_CMD_START_CONSUM);
} }