enh: add connection type for tmq

example/src/tmq.c

@@ -144,6 +144,7 @@ void tmq_commit_cb_print(tmq_t* tmq, tmq_resp_err_t resp, tmq_topic_vgroup_list_
 }
 
 tmq_t* build_consumer() {
+#if 0
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   assert(pConn != NULL);
 
@@ -152,11 +153,15 @@ tmq_t* build_consumer() {
     printf("error in use db, reason:%s\n", taos_errstr(pRes));
   }
   taos_free_result(pRes);
+#endif
 
   tmq_conf_t* conf = tmq_conf_new();
   tmq_conf_set(conf, "group.id", "tg2");
+  tmq_conf_set(conf, "td.connect.user", "root");
+  tmq_conf_set(conf, "td.connect.pass", "taosdata");
+  tmq_conf_set(conf, "td.connect.db", "abc1");
   tmq_conf_set_offset_commit_cb(conf, tmq_commit_cb_print);
-  tmq_t* tmq = tmq_consumer_new(pConn, conf, NULL, 0);
+  tmq_t* tmq = tmq_consumer_new1(conf, NULL, 0);
   return tmq;
 }
 

include/client/taos.h

@@ -247,10 +247,10 @@ DLL_EXPORT tmq_list_t *tmq_list_new();
 DLL_EXPORT int32_t     tmq_list_append(tmq_list_t *, const char *);
 DLL_EXPORT void        tmq_list_destroy(tmq_list_t *);
 
-// will be removed in 3.0
+#if 1
 DLL_EXPORT tmq_t *tmq_consumer_new(void *conn, tmq_conf_t *conf, char *errstr, int32_t errstrLen);
+#endif
 
-// will replace last one
 DLL_EXPORT tmq_t *tmq_consumer_new1(tmq_conf_t *conf, char *errstr, int32_t errstrLen);
 
 DLL_EXPORT const char *tmq_err2str(tmq_resp_err_t);

include/common/tmsg.h

@@ -332,6 +332,7 @@ int32_t taosEncodeSEpSet(void** buf, const SEpSet* pEp);
 void*   taosDecodeSEpSet(void* buf, SEpSet* pEp);
 
 typedef struct {
+  int8_t  connType;
   int32_t pid;
   char    app[TSDB_APP_NAME_LEN];
   char    db[TSDB_DB_NAME_LEN];
@@ -346,6 +347,7 @@ typedef struct {
   int64_t clusterId;
   int32_t connId;
   int8_t  superUser;
+  int8_t  connType;
   SEpSet  epSet;
   char    sVersion[128];
 } SConnectRsp;

source/client/inc/clientInt.h

@@ -45,6 +45,11 @@ extern "C" {
 
 #define HEARTBEAT_INTERVAL 1500  // ms
 
+enum {
+  CONN_TYPE__QUERY = 1,
+  CONN_TYPE__TMQ,
+};
+
 typedef struct SAppInstInfo SAppInstInfo;
 
 typedef struct {
@@ -132,9 +137,9 @@ typedef struct STscObj {
   char          pass[TSDB_PASSWORD_LEN];
   char          db[TSDB_DB_FNAME_LEN];
   char          ver[128];
+  int8_t        connType;
   int32_t       acctId;
   uint32_t      connId;
-  int32_t       connType;
   uint64_t      id;         // ref ID returned by taosAddRef
   TdThreadMutex mutex;      // used to protect the operation on db
   int32_t       numOfReqs;  // number of sqlObj bound to this connection
@@ -272,7 +277,7 @@ void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet);
 void initMsgHandleFp();
 
 TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
-                            uint16_t port);
+                            uint16_t port, int connType);
 
 int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery);
 int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList);

source/client/src/clientHb.c

@@ -23,6 +23,8 @@ static SClientHbMgr clientHbMgr = {0};
 static int32_t hbCreateThread();
 static void    hbStopThread();
 
+static int32_t hbMqHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req) { return 0; }
+
 static int32_t hbMqHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { return 0; }
 
 static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
@@ -297,11 +299,10 @@ int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t hbMqHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req) { return 0; }
-
 void hbMgrInitMqHbHandle() {
   clientHbMgr.reqHandle[HEARTBEAT_TYPE_QUERY] = hbQueryHbReqHandle;
   clientHbMgr.reqHandle[HEARTBEAT_TYPE_MQ] = hbMqHbReqHandle;
+
   clientHbMgr.rspHandle[HEARTBEAT_TYPE_QUERY] = hbQueryHbRspHandle;
   clientHbMgr.rspHandle[HEARTBEAT_TYPE_MQ] = hbMqHbRspHandle;
 }
@@ -568,7 +569,7 @@ int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, SHbConnInfo *
 int hbRegisterConn(SAppHbMgr *pAppHbMgr, int32_t connId, int64_t clusterId, int32_t hbType) {
   SClientHbKey connKey = {
       .connId = connId,
-      .hbType = HEARTBEAT_TYPE_QUERY,
+      .hbType = hbType,
   };
   SHbConnInfo info = {0};
 
@@ -578,16 +579,14 @@ int hbRegisterConn(SAppHbMgr *pAppHbMgr, int32_t connId, int64_t clusterId, int3
       *pClusterId = clusterId;
 
       info.param = pClusterId;
-      break;
+      return hbRegisterConnImpl(pAppHbMgr, connKey, &info);
     }
     case HEARTBEAT_TYPE_MQ: {
-      break;
+      return 0;
     }
     default:
-      break;
+      return 0;
   }
-
-  return hbRegisterConnImpl(pAppHbMgr, connKey, &info);
 }
 
 void hbDeregisterConn(SAppHbMgr *pAppHbMgr, SClientHbKey connKey) {

source/client/src/clientImpl.c

@@ -11,7 +11,7 @@
 #include "tref.h"
 
 static int32_t       initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet);
-static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest);
+static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest, int8_t connType);
 static void          destroySendMsgInfo(SMsgSendInfo* pMsgBody);
 
 static bool stringLengthCheck(const char* str, size_t maxsize) {
@@ -40,10 +40,10 @@ static char* getClusterKey(const char* user, const char* auth, const char* ip, i
 }
 
 static STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __taos_async_fn_t fp, void* param,
-                                SAppInstInfo* pAppInfo);
+                                SAppInstInfo* pAppInfo, int connType);
 
 TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
-                            uint16_t port) {
+                            uint16_t port, int connType) {
   if (taos_init() != TSDB_CODE_SUCCESS) {
     return NULL;
   }
@@ -111,7 +111,7 @@ TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass,
   taosThreadMutexUnlock(&appInfo.mutex);
 
   taosMemoryFreeClear(key);
-  return taosConnectImpl(user, &secretEncrypt[0], localDb, NULL, NULL, *pInst);
+  return taosConnectImpl(user, &secretEncrypt[0], localDb, NULL, NULL, *pInst, connType);
 }
 
 int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj** pRequest) {
@@ -425,7 +425,7 @@ int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSe
 }
 
 STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __taos_async_fn_t fp, void* param,
-                         SAppInstInfo* pAppInfo) {
+                         SAppInstInfo* pAppInfo, int connType) {
   STscObj* pTscObj = createTscObj(user, auth, db, pAppInfo);
   if (NULL == pTscObj) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
@@ -439,7 +439,7 @@ STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __t
     return NULL;
   }
 
-  SMsgSendInfo* body = buildConnectMsg(pRequest);
+  SMsgSendInfo* body = buildConnectMsg(pRequest, connType);
 
   int64_t transporterId = 0;
   asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
@@ -462,7 +462,7 @@ STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __t
   return pTscObj;
 }
 
-static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest) {
+static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest, int8_t connType) {
   SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
   if (pMsgSendInfo == NULL) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
@@ -485,6 +485,7 @@ static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest) {
   }
   taosMemoryFreeClear(db);
 
+  connectReq.connType = connType;
   connectReq.pid = htonl(appInfo.pid);
   connectReq.startTime = htobe64(appInfo.startTime);
   tstrncpy(connectReq.app, appInfo.appName, sizeof(connectReq.app));
@@ -570,7 +571,7 @@ TAOS* taos_connect_auth(const char* ip, const char* user, const char* auth, cons
     return NULL;
   }
 
-  return taos_connect_internal(ip, user, NULL, auth, db, port);
+  return taos_connect_internal(ip, user, NULL, auth, db, port, CONN_TYPE__QUERY);
 }
 
 TAOS* taos_connect_l(const char* ip, int ipLen, const char* user, int userLen, const char* pass, int passLen,

source/client/src/clientMain.c

@@ -87,7 +87,7 @@ TAOS *taos_connect(const char *ip, const char *user, const char *pass, const cha
     pass = TSDB_DEFAULT_PASS;
   }
 
-  return taos_connect_internal(ip, user, pass, NULL, db, port);
+  return taos_connect_internal(ip, user, pass, NULL, db, port, CONN_TYPE__QUERY);
 }
 
 void taos_close(TAOS *taos) {
@@ -124,9 +124,11 @@ const char *taos_errstr(TAOS_RES *res) {
 }
 
 void taos_free_result(TAOS_RES *res) {
+  if (TD_RES_QUERY(res)) {
     SRequestObj *pRequest = (SRequestObj *)res;
     destroyRequest(pRequest);
   }
+}
 
 int taos_field_count(TAOS_RES *res) {
   if (res == NULL) {

source/client/src/clientMsgHandler.c

@@ -13,13 +13,13 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "os.h"
+#include "catalog.h"
-#include "tdef.h"
-#include "tname.h"
 #include "clientInt.h"
 #include "clientLog.h"
-#include "catalog.h"
+#include "os.h"
 #include "query.h"
+#include "tdef.h"
+#include "tname.h"
 
 int32_t (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code);
 
@@ -69,9 +69,9 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
   pTscObj->pAppInfo->clusterId = connectRsp.clusterId;
   atomic_add_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
 
-  pTscObj->connType = HEARTBEAT_TYPE_QUERY;
+  pTscObj->connType = connectRsp.connType;
 
-  hbRegisterConn(pTscObj->pAppInfo->pAppHbMgr, connectRsp.connId, connectRsp.clusterId, HEARTBEAT_TYPE_QUERY);
+  hbRegisterConn(pTscObj->pAppInfo->pAppHbMgr, connectRsp.connId, connectRsp.clusterId, connectRsp.connType);
 
   //  pRequest->body.resInfo.pRspMsg = pMsg->pData;
   tscDebug("0x%" PRIx64 " clusterId:%" PRId64 ", totalConn:%" PRId64, pRequest->requestId, connectRsp.clusterId,
@@ -119,7 +119,9 @@ SMsgSendInfo* buildMsgInfoImpl(SRequestObj *pRequest) {
     pMsgSendInfo->msgInfo = pRequest->body.requestMsg;
   }
 
-  pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL)? genericRspCallback:handleRequestRspFp[TMSG_INDEX(pRequest->type)];
+  pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL)
+                         ? genericRspCallback
+                         : handleRequestRspFp[TMSG_INDEX(pRequest->type)];
   return pMsgSendInfo;
 }
 
@@ -256,7 +258,8 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
     if (usedbRsp.vgVersion >= 0) {
       int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
       if (code != TSDB_CODE_SUCCESS) {
-        tscWarn("catalogGetHandle failed, clusterId:%"PRIx64", error:%s", pRequest->pTscObj->pAppInfo->clusterId, tstrerror(code));
+        tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pRequest->pTscObj->pAppInfo->clusterId,
+                tstrerror(code));
       } else {
         catalogRemoveDB(pCatalog, usedbRsp.db, usedbRsp.uid);
       }
@@ -292,7 +295,8 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
 
     int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
     if (code != TSDB_CODE_SUCCESS) {
-      tscWarn("catalogGetHandle failed, clusterId:%"PRIx64", error:%s", pRequest->pTscObj->pAppInfo->clusterId, tstrerror(code));
+      tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pRequest->pTscObj->pAppInfo->clusterId,
+              tstrerror(code));
     } else {
       catalogUpdateDBVgInfo(pCatalog, output.db, output.dbId, output.dbVgroup);
     }

source/client/src/tmq.c

@@ -357,7 +357,15 @@ tmq_t* tmq_consumer_new1(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
   if (pTmq == NULL) {
     return NULL;
   }
-  pTmq->pTscObj = taos_connect(conf->ip, conf->user, conf->pass, conf->db, conf->port);
+  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);
+  ASSERT(conf->db);
+
+  pTmq->pTscObj = taos_connect_internal(conf->ip, user, pass, NULL, conf->db, conf->port, CONN_TYPE__TMQ);
+  if (pTmq->pTscObj == NULL) return NULL;
 
   pTmq->inWaiting = 0;
   pTmq->status = 0;

source/common/src/tmsg.c

@@ -2532,6 +2532,7 @@ int32_t tSerializeSConnectReq(void *buf, int32_t bufLen, SConnectReq *pReq) {
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
 
   if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->connType) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->pid) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->app) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->db) < 0) return -1;
@@ -2548,6 +2549,7 @@ int32_t tDeserializeSConnectReq(void *buf, int32_t bufLen, SConnectReq *pReq) {
   tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
 
   if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeI8(&decoder, &pReq->connType) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->pid) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->app) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->db) < 0) return -1;
@@ -2567,6 +2569,7 @@ int32_t tSerializeSConnectRsp(void *buf, int32_t bufLen, SConnectRsp *pRsp) {
   if (tEncodeI64(&encoder, pRsp->clusterId) < 0) return -1;
   if (tEncodeI32(&encoder, pRsp->connId) < 0) return -1;
   if (tEncodeI8(&encoder, pRsp->superUser) < 0) return -1;
+  if (tEncodeI8(&encoder, pRsp->connType) < 0) return -1;
   if (tEncodeSEpSet(&encoder, &pRsp->epSet) < 0) return -1;
   if (tEncodeCStr(&encoder, pRsp->sVersion) < 0) return -1;
   tEndEncode(&encoder);
@@ -2585,6 +2588,7 @@ int32_t tDeserializeSConnectRsp(void *buf, int32_t bufLen, SConnectRsp *pRsp) {
   if (tDecodeI64(&decoder, &pRsp->clusterId) < 0) return -1;
   if (tDecodeI32(&decoder, &pRsp->connId) < 0) return -1;
   if (tDecodeI8(&decoder, &pRsp->superUser) < 0) return -1;
+  if (tDecodeI8(&decoder, &pRsp->connType) < 0) return -1;
   if (tDecodeSEpSet(&decoder, &pRsp->epSet) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pRsp->sVersion) < 0) return -1;
   tEndDecode(&decoder);

source/dnode/mnode/impl/src/mndProfile.c

@@ -30,6 +30,7 @@
 
 typedef struct {
   int32_t     id;
+  int8_t      connType;
   char        user[TSDB_USER_LEN];
   char        app[TSDB_APP_NAME_LEN];  // app name that invokes taosc
   int64_t     appStartTimeMs;          // app start time
@@ -44,8 +45,8 @@ typedef struct {
   SQueryDesc *pQueries;
 } SConnObj;
 
-static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, uint32_t ip, uint16_t port, int32_t pid,
+static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType, uint32_t ip, uint16_t port,
-                               const char *app, int64_t startTime);
+                               int32_t pid, const char *app, int64_t startTime);
 static void      mndFreeConn(SConnObj *pConn);
 static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId);
 static void      mndReleaseConn(SMnode *pMnode, SConnObj *pConn);
@@ -93,8 +94,8 @@ void mndCleanupProfile(SMnode *pMnode) {
   }
 }
 
-static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, uint32_t ip, uint16_t port, int32_t pid,
+static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType, uint32_t ip, uint16_t port,
-                               const char *app, int64_t startTime) {
+                               int32_t pid, const char *app, int64_t startTime) {
   SProfileMgmt *pMgmt = &pMnode->profileMgmt;
 
   int32_t connId = atomic_add_fetch_32(&pMgmt->connId, 1);
@@ -102,6 +103,7 @@ static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, uint32_t ip, ui
   if (startTime == 0) startTime = taosGetTimestampMs();
 
   SConnObj connObj = {.id = connId,
+                      .connType = connType,
                       .appStartTimeMs = startTime,
                       .pid = pid,
                       .ip = ip,
@@ -210,8 +212,8 @@ static int32_t mndProcessConnectReq(SNodeMsg *pReq) {
     }
   }
 
-  pConn =
+  pConn = mndCreateConn(pMnode, pReq->user, connReq.connType, pReq->clientIp, pReq->clientPort, connReq.pid,
-      mndCreateConn(pMnode, pReq->user, pReq->clientIp, pReq->clientPort, connReq.pid, connReq.app, connReq.startTime);
+                        connReq.app, connReq.startTime);
   if (pConn == NULL) {
     mError("user:%s, failed to login from %s while create connection since %s", pReq->user, ip, terrstr());
     goto CONN_OVER;
@@ -222,6 +224,7 @@ static int32_t mndProcessConnectReq(SNodeMsg *pReq) {
   connectRsp.superUser = pUser->superUser;
   connectRsp.clusterId = pMnode->clusterId;
   connectRsp.connId = pConn->id;
+  connectRsp.connType = connReq.connType;
 
   snprintf(connectRsp.sVersion, sizeof(connectRsp.sVersion), "ver:%s\nbuild:%s\ngitinfo:%s", version, buildinfo,
            gitinfo);
@@ -343,7 +346,6 @@ static int32_t mndProcessHeartBeatReq(SNodeMsg *pReq) {
     return -1;
   }
 
-  
   SClientHbBatchRsp batchRsp = {0};
   batchRsp.rsps = taosArrayInit(0, sizeof(SClientHbRsp));
 

tests/test/c/tmqDemo.c

@@ -13,17 +13,15 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-// clang-format off
-
 #include <assert.h>
+#include <dirent.h>
 #include <stdio.h>
+#include <stdlib.h>
 #include <string.h>
-#include <time.h>
 #include <sys/stat.h>
 #include <sys/types.h>
+#include <time.h>
 #include <unistd.h>
-#include <stdlib.h>
-#include <dirent.h>
 
 #include "taos.h"
 #include "taoserror.h"
@@ -36,12 +34,7 @@
 #define MAX_SQL_STR_LEN (1024 * 1024)
 #define MAX_ROW_STR_LEN (16 * 1024)
 
-enum _RUN_MODE {
+enum _RUN_MODE { TMQ_RUN_INSERT_AND_CONSUME, TMQ_RUN_ONLY_INSERT, TMQ_RUN_ONLY_CONSUME, TMQ_RUN_MODE_BUTT };
-    TMQ_RUN_INSERT_AND_CONSUME,
-    TMQ_RUN_ONLY_INSERT,
-    TMQ_RUN_ONLY_CONSUME,
-    TMQ_RUN_MODE_BUTT
-};
 
 typedef struct {
   char    dbName[32];
@@ -126,10 +119,8 @@ static void printHelp() {
 }
 
 void parseArgument(int32_t argc, char* argv[]) {
-
   g_stConfInfo.startTimestamp = 1640966400000;  // 2020-01-01 00:00:00.000
 
-
   for (int32_t i = 1; i < argc; i++) {
     if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
       printHelp();
@@ -198,8 +189,7 @@ static int  running = 1;
 /*static void msg_process(tmq_message_t* message) { tmqShowMsg(message); }*/
 
 // calc dir size (not include itself 4096Byte)
-int64_t getDirectorySize(char *dir)
+int64_t getDirectorySize(char* dir) {
-{
   TdDirPtr      pDir;
   TdDirEntryPtr pDirEntry;
   int64_t       totalSize = 0;
@@ -236,7 +226,6 @@ int64_t getDirectorySize(char *dir)
   return totalSize;
 }
 
-
 int queryDB(TAOS* taos, char* command) {
   TAOS_RES* pRes = taos_query(taos, command);
   int       code = taos_errno(pRes);
@@ -327,6 +316,7 @@ int32_t init_env() {
 }
 
 tmq_t* build_consumer() {
+#if 0
   char sqlStr[1024] = {0};
   
   TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0);
@@ -338,10 +328,16 @@ tmq_t* build_consumer() {
     printf("error in use db, reason:%s\n", taos_errstr(pRes));
   }
   taos_free_result(pRes);
+#endif
 
   tmq_conf_t* conf = tmq_conf_new();
   tmq_conf_set(conf, "group.id", "tg2");
-  tmq_t* tmq = tmq_consumer_new(pConn, conf, NULL, 0);
+  tmq_conf_set(conf, "td.connect.user", "root");
+  tmq_conf_set(conf, "td.connect.pass", "taosdata");
+  tmq_conf_set(conf, "td.connect.db", g_stConfInfo.dbName);
+  tmq_t* tmq = tmq_consumer_new1(conf, NULL, 0);
+  assert(tmq);
+  tmq_conf_destroy(conf);
   return tmq;
 }
 
@@ -417,7 +413,9 @@ void perf_loop(tmq_t* tmq, tmq_list_t* topics, int32_t totalMsgs, int64_t walLog
   } else {
     printf("{consume success: %d}", totalMsgs);
   }
-  taosFprintfFile(g_fp, "|%10d    |   %10.3f    |  %8.2f  |  %10.2f|    %10.2f    |\n", batchCnt, consumeTime, (double)batchCnt / consumeTime, (double)walLogSize / (1024 * 1024.0) / consumeTime, (double)walLogSize / 1024.0 / batchCnt);
+  taosFprintfFile(g_fp, "|%10d    |   %10.3f    |  %8.2f  |  %10.2f|    %10.2f    |\n", batchCnt, consumeTime,
+                  (double)batchCnt / consumeTime, (double)walLogSize / (1024 * 1024.0) / consumeTime,
+                  (double)walLogSize / 1024.0 / batchCnt);
 
   err = tmq_consumer_close(tmq);
   if (err) {
@@ -492,7 +490,6 @@ int32_t syncWriteData() {
   return totalMsgs;
 }
 
-
 // sync insertion
 int32_t syncWriteDataByRatio() {
   TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0);
@@ -581,14 +578,16 @@ int32_t syncWriteDataByRatio() {
       totalMsgs++;
     }
   }
-  pPrint("expect insert rows: T1[%d] T2[%d], actual insert rows: T1[%d] T2[%d]\n", g_stConfInfo.totalRowsOfPerTbl, g_stConfInfo.totalRowsOfT2, insertedOfT1, insertedOfT2);
+  pPrint("expect insert rows: T1[%d] T2[%d], actual insert rows: T1[%d] T2[%d]\n", g_stConfInfo.totalRowsOfPerTbl,
+         g_stConfInfo.totalRowsOfT2, insertedOfT1, insertedOfT2);
   taosMemoryFreeClear(buffer);
   return totalMsgs;
 }
 
 void printParaIntoFile() {
   // FILE *fp = fopen(g_stConfInfo.resultFileName, "a");
-  TdFilePtr pFile = taosOpenFile(g_stConfInfo.resultFileName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND | TD_FILE_STREAM);
+  TdFilePtr pFile =
+      taosOpenFile(g_stConfInfo.resultFileName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND | TD_FILE_STREAM);
   if (NULL == pFile) {
     fprintf(stderr, "Failed to open %s for save result\n", g_stConfInfo.resultFileName);
     exit - 1;
@@ -615,8 +614,13 @@ void printParaIntoFile() {
   taosFprintfFile(pFile, "# Test time:                %d-%02d-%02d %02d:%02d:%02d\n", tm.tm_year + 1900, tm.tm_mon + 1,
                   tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec);
   taosFprintfFile(pFile, "###################################################################\n");
-  taosFprintfFile(pFile, "|-------------------------------insert info-----------------------------|--------------------------------consume info---------------------------------|\n");
+  taosFprintfFile(pFile,
-  taosFprintfFile(pFile, "|batch size| insert msgs | insert time(s) |   msgs/s   | walLogSize(MB) | consume msgs | consume time(s) |   msgs/s   |    MB/s    | avg msg size(KB) |\n");
+                  "|-------------------------------insert "
+                  "info-----------------------------|--------------------------------consume "
+                  "info---------------------------------|\n");
+  taosFprintfFile(pFile,
+                  "|batch size| insert msgs | insert time(s) |   msgs/s   | walLogSize(MB) | consume msgs | consume "
+                  "time(s) |   msgs/s   |    MB/s    | avg msg size(KB) |\n");
   taosFprintfFile(g_fp, "|%10d", g_stConfInfo.batchNumOfRow);
 }
 
@@ -636,7 +640,6 @@ int main(int32_t argc, char *argv[]) {
   int32_t totalMsgs = 0;
 
   if (g_stConfInfo.runMode != TMQ_RUN_ONLY_CONSUME) {
-  	
     int64_t startTs = taosGetTimestampUs();
     if (1 == g_stConfInfo.ratio) {
       totalMsgs = syncWriteData();
@@ -662,7 +665,6 @@ int main(int32_t argc, char *argv[]) {
     float rowsSpeed = totalRows / seconds;
     float msgsSpeed = totalMsgs / seconds;
 
-
     if ((0 == g_stConfInfo.simCase) && (strlen(g_stConfInfo.vnodeWalPath))) {
       walLogSize = getDirectorySize(g_stConfInfo.vnodeWalPath);
       if (walLogSize <= 0) {
@@ -674,9 +676,11 @@ int main(int32_t argc, char *argv[]) {
     }
 
     if (0 == g_stConfInfo.simCase) {
-	  pPrint("insert result: %d rows, %d msgs, time:%.3f sec, speed:%.1f rows/second, %.1f msgs/second\n", totalRows, totalMsgs, seconds, rowsSpeed, msgsSpeed);
+      pPrint("insert result: %d rows, %d msgs, time:%.3f sec, speed:%.1f rows/second, %.1f msgs/second\n", totalRows,
+             totalMsgs, seconds, rowsSpeed, msgsSpeed);
     }
-	taosFprintfFile(g_fp, "|%10d   |   %10.3f   |  %8.2f  |   %10.3f   ", totalMsgs, seconds, msgsSpeed, (double)walLogSize/(1024 * 1024.0));
+    taosFprintfFile(g_fp, "|%10d   |   %10.3f   |  %8.2f  |   %10.3f   ", totalMsgs, seconds, msgsSpeed,
+                    (double)walLogSize / (1024 * 1024.0));
   }
 
   if (g_stConfInfo.runMode == TMQ_RUN_ONLY_INSERT) {

tests/test/c/tmqSim.c

@@ -227,7 +227,8 @@ void parseInputString() {
       char* ret = strchr(pstr, ch);
       memcpy(g_stConfInfo.key1[g_stConfInfo.numOfKey1], pstr, ret - pstr);
       strcpy(g_stConfInfo.value1[g_stConfInfo.numOfKey1], ret + 1);
-	//printf("key: %s, value: %s\n", g_stConfInfo.key[g_stConfInfo.numOfKey], g_stConfInfo.value[g_stConfInfo.numOfKey]);
+      // printf("key: %s, value: %s\n", g_stConfInfo.key[g_stConfInfo.numOfKey],
+      // g_stConfInfo.value[g_stConfInfo.numOfKey]);
       g_stConfInfo.numOfKey1++;
     }
 
@@ -235,7 +236,6 @@ void parseInputString() {
   }
 }
 
-
 static int running = 1;
 /*static void msg_process(tmq_message_t* message) { tmqShowMsg(message); }*/
 
@@ -253,6 +253,7 @@ int queryDB(TAOS* taos, char* command) {
 }
 
 tmq_t* build_consumer() {
+#if 0
   char sqlStr[1024] = {0};
 
   TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0);
@@ -266,13 +267,19 @@ tmq_t* build_consumer() {
     exit(-1);
   }
   taos_free_result(pRes);
+#endif
 
   tmq_conf_t* conf = tmq_conf_new();
   // tmq_conf_set(conf, "group.id", "tg2");
   for (int32_t i = 0; i < g_stConfInfo.numOfKey; i++) {
     tmq_conf_set(conf, g_stConfInfo.key[i], g_stConfInfo.value[i]);
   }
-  tmq_t* tmq = tmq_consumer_new(pConn, conf, NULL, 0);
+  tmq_conf_set(conf, "td.connect.user", "root");
+  tmq_conf_set(conf, "td.connect.pass", "taosdata");
+  tmq_conf_set(conf, "td.connect.db", g_stConfInfo.dbName);
+  tmq_t* tmq = tmq_consumer_new1(conf, NULL, 0);
+  assert(tmq);
+  tmq_conf_destroy(conf);
   return tmq;
 }
 
@@ -285,8 +292,8 @@ tmq_list_t* build_topic_list() {
   return topic_list;
 }
 
-
 tmq_t* build_consumer_x() {
+#if 0
   char sqlStr[1024] = {0};
 
   TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0);
@@ -300,13 +307,19 @@ tmq_t* build_consumer_x() {
     exit(-1);
   }
   taos_free_result(pRes);
+#endif
 
   tmq_conf_t* conf = tmq_conf_new();
   // tmq_conf_set(conf, "group.id", "tg2");
   for (int32_t i = 0; i < g_stConfInfo.numOfKey1; i++) {
     tmq_conf_set(conf, g_stConfInfo.key1[i], g_stConfInfo.value1[i]);
   }
-  tmq_t* tmq = tmq_consumer_new(pConn, conf, NULL, 0);
+  tmq_conf_set(conf, "td.connect.user", "root");
+  tmq_conf_set(conf, "td.connect.pass", "taosdata");
+  tmq_conf_set(conf, "td.connect.db", g_stConfInfo.dbName);
+  tmq_t* tmq = tmq_consumer_new1(conf, NULL, 0);
+  assert(tmq);
+  tmq_conf_destroy(conf);
   return tmq;
 }
 
@@ -400,7 +413,6 @@ int32_t parallel_consume(tmq_t* tmq, int threadLable) {
   return totalMsgs;
 }
 
-
 void* threadFunc(void* param) {
   int32_t totalMsgs = 0;
 
@@ -434,7 +446,6 @@ void *threadFunc(void *param) {
   return NULL;
 }
 
-
 int main(int32_t argc, char* argv[]) {
   parseArgument(argc, argv);
   parseInputString();
@@ -510,10 +521,10 @@ int main(int32_t argc, char* argv[]) {
         printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt);
       }
     } else if (4 == g_stConfInfo.checkMode) {
-      if (((totalMsgs == 0) && (pInfo->consumeMsgCnt == 3 * g_stConfInfo.consumeMsgCnt)) 
+      if (((totalMsgs == 0) && (pInfo->consumeMsgCnt == 3 * g_stConfInfo.consumeMsgCnt)) ||
-	  	|| ((pInfo->consumeMsgCnt == 0) && (totalMsgs == 3 * g_stConfInfo.consumeMsgCnt)) 
+          ((pInfo->consumeMsgCnt == 0) && (totalMsgs == 3 * g_stConfInfo.consumeMsgCnt)) ||
-	  	|| ((pInfo->consumeMsgCnt == g_stConfInfo.consumeMsgCnt) && (totalMsgs == 2 * g_stConfInfo.consumeMsgCnt))
+          ((pInfo->consumeMsgCnt == g_stConfInfo.consumeMsgCnt) && (totalMsgs == 2 * g_stConfInfo.consumeMsgCnt)) ||
-	  	|| ((pInfo->consumeMsgCnt == 2 * g_stConfInfo.consumeMsgCnt) && (totalMsgs == g_stConfInfo.consumeMsgCnt))) {
+          ((pInfo->consumeMsgCnt == 2 * g_stConfInfo.consumeMsgCnt) && (totalMsgs == g_stConfInfo.consumeMsgCnt))) {
         printf("success");
       } else {
         printf("fail, consumer msg cnt: %d, %d", totalMsgs, pInfo->consumeMsgCnt);