add some msg for tq (#8241)

include/common/taosmsg.h

@@ -41,6 +41,10 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SUBMIT, "submit" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_QUERY, "query" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_FETCH, "fetch" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_UPDATE_TAG_VAL, "update-tag-val" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_CONNECT, "mq-connect" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_CONSUME, "mq-consume" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_ACK, "mq-ack" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_RESET, "mq-reset" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY1, "dummy1" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY2, "dummy2" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY3, "dummy3" )
@@ -113,7 +117,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_NETWORK_TEST, "nettest" )
 // message for topic
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_CREATE_TP, "create-tp" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_DROP_TP, "drop-tp" )	  
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_USE_TP, "use-tp" )	 
+//TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_USE_TP, "use-tp" )	 
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_ALTER_TP, "alter-tp" )
 
 #ifndef TAOS_MESSAGE_C

include/server/vnode/tq/tq.h

@@ -22,6 +22,56 @@
 extern "C" {
 #endif
 
+typedef struct tmqMsgHead {
+  int32_t headLen;
+  int32_t msgVer;
+  int64_t cgId;
+  int32_t topicLen;
+  char topic[];
+} tmqMsgHead;
+
+//TODO: put msgs into common
+typedef struct tmqConnectReq {
+  tmqMsgHead head;
+
+} tmqConnectReq;
+
+typedef struct tmqConnectResp {
+
+} tmqConnectResp;
+
+typedef struct tmqDisconnectReq {
+
+} tmqDisconnectReq;
+
+typedef struct tmqDisconnectResp {
+
+} tmqDiconnectResp;
+
+typedef struct tmqConsumeReq {
+
+} tmqConsumeReq;
+
+typedef struct tmqConsumeResp {
+
+} tmqConsumeResp;
+
+typedef struct tmqSubscribeReq {
+
+} tmqSubscribeReq;
+
+typedef struct tmqSubscribeResp {
+
+} tmqSubscribeResp;
+
+typedef struct tmqHeartbeatReq {
+
+} tmqHeartbeatReq;
+
+typedef struct tmqHeartbeatResp {
+
+} tmqHeartbeatResp;
+
 typedef struct tqTopicVhandle {
   //name
   //
@@ -29,7 +79,7 @@ typedef struct tqTopicVhandle {
   //
   //callback for mnode
   //
-} tqTopic;
+} tqTopicVhandle;
 
 typedef struct STQ {
   //the set for topics
@@ -50,7 +100,7 @@ int tqPushMsg(STQ*, void* msg, int64_t version);
 int tqCommit(STQ*);
 
 //void* will be replace by a msg type
-int tqHandleMsg(STQ*, void* msg);
+int tqHandleConsumeMsg(STQ*, tmqConsumeReq* msg);
 
 #ifdef __cplusplus
 }

source/server/vnode/tq/inc/tqInt.h

@@ -26,14 +26,15 @@ extern "C" {
 
 typedef struct tqBufferItem {
   int64_t offset;
-  void *content;
+  void* executor;
+  void* content;
 } tqBufferItem;
 
 
 typedef struct tqGroupHandle {
-  char* topic;
+  char* topic; //c style, end with '\0'
-  void* ahandle;
   int64_t cgId;
+  void* ahandle;
   int64_t consumeOffset;
   int32_t head;
   int32_t tail;

source/server/vnode/tq/src/tq.c

@@ -77,7 +77,7 @@ int tqCommit(STQ* pTq) {
   return 0;
 }
 
-int tqHandleMsg(STQ* pTq, void*msg) {
+int tqHandleConsumeMsg(STQ* pTq, tmqConsumeReq* msg) {
   //parse msg and extract topic and cgId
   //lookup handle
   //confirm message and send to consumer