add tq header

include/client/consumer/consumer.h

@@ -16,6 +16,10 @@
 #ifndef _TD_CONSUMER_H_
 #define _TD_CONSUMER_H_
 
+#include "tlist.h"
+#include "tarray.h"
+#include "hash.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -32,16 +36,15 @@ extern "C" {
   struct tmq_resp_err_t;
   typedef struct tmq_resp_err_t tmq_resp_err_t;
 
-  //topic list
-  //resouces are supposed to be free by users by calling tmq_list_destroy
-  struct tmq_topic_list_t;
-  typedef struct tmq_topic_list_t tmq_topic_list_t;
-  int32_t tmq_topic_list_add(tmq_topic_list_t*, const char*);
-  void    tmq_topic_list_destroy(tmq_topic_list_t*);
+  struct tmq_message_t;
+  typedef struct tmq_message_t tmq_message_t;
+
+  struct tmq_col_batch_t;
+  typedef struct tmq_col_batch_t tmq_col_batch_t;
 
   //get content of message
-  tmq_col_batch_t *tmq_get_msg_col_by_idx(tmq_message_t*, int32_t);
-  tmq_col_batch_t *tmq_get_msg_col_by_name(tmq_message_t*, const char*);
+  tmq_col_batch_t* tmq_get_msg_col_by_idx(tmq_message_t*, int32_t col_id);
+  tmq_col_batch_t* tmq_get_msg_col_by_name(tmq_message_t*, const char*);
 
   //consumer config
   int32_t tmq_conf_set(tmq_consumer_config_t* , const char* config_key, const char* config_value, char* errstr, int32_t errstr_cap);
@@ -51,11 +54,12 @@ extern "C" {
   tmq_consumer_t* tmq_consumer_new(tmq_consumer_config_t* , char* errstr, int32_t errstr_cap);
 
   //subscribe
-  tmq_resp_err_t tmq_subscribe(tmq_consumer_t*, const tmq_topic_list_t*);
+  tmq_resp_err_t tmq_subscribe(tmq_consumer_t*, const SList*);
+  tmq_resp_err_t tmq_unsubscribe(tmq_consumer_t*);
 
   //consume
   //resouces are supposed to be free by users by calling tmq_message_destroy
-  tmq_message_t tmq_consume_poll(tmq_consumer_t*, int64_t blocking_time);
+  tmq_message_t* tmq_consume_poll(tmq_consumer_t*, int64_t blocking_time);
 
   //destroy message and free memory
   void tmq_message_destroy(tmq_message_t*);

include/server/vnode/tq/tq.h

@@ -16,13 +16,29 @@
 #ifndef _TD_TQ_H_
 #define _TD_TQ_H_
 
+#include "os.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
 
 typedef struct STQ STQ;
 
-int tqPushMsg(void *);
+STQ* tqInit();
+void tqCleanUp(STQ* pTQ);
+
+//create persistent storage for meta info such as consuming offset
+//return value > 0: cgId
+//return value < 0: error code
+int tqCreateGroup(STQ *pTQ);
+//create ring buffer in memory and load consuming offset
+int tqOpenGroup(STQ* pTQ, int cgId);
+//destroy ring buffer and persist consuming offset
+int tqCloseGroup(STQ *pTQ, int cgId);
+//delete persistent storage for meta info
+int tqDropGroup(STQ *pTQ);
+
+int tqPushMsg(STQ *pTQ, void *, int64_t version);
 int tqCommit(STQ *pTQ);
 
 #ifdef __cplusplus

source/client/consumer/consumer.c

@@ -12,3 +12,5 @@
  * 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 "consumer.h"

source/server/vnode/tq/CMakeLists.txt

@@ -3,5 +3,10 @@ add_library(tq ${TQ_SRC})
 target_include_directories(
     tq
     PUBLIC "${CMAKE_SOURCE_DIR}/include/server/vnode/tq"
+    PRIVATE "${CMAKE_SOURCE_DIR}/include/os"
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
+
+target_link_libraries(
+    os
+)

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

@@ -20,6 +20,9 @@
 extern "C" {
 #endif
 
+//implement the array index
+//implement the ring buffer
+
 #ifdef __cplusplus
 }
 #endif

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

@@ -15,5 +15,12 @@
 
 #include "tq.h"
 
-int tqPushMsg(void * p) {return 0;}
-int tqCommit(STQ *pTQ) {return 0;}
+int tqPushMsg(STQ *pTQ, void * p, int64_t version) {
+  //add reference
+  //
+  return 0;
+}
+
+int tqCommit(STQ *pTQ) {
+  return 0;
+}