refact

source/dnode/vnode/impl/inc/vnodeCommit.h

@@ -16,13 +16,14 @@
 #ifndef _TD_VNODE_COMMIT_H_
 #define _TD_VNODE_COMMIT_H_
 
-#include "vnodeInt.h"
+#include "vnode.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-int vnodeAsyncCommit(SVnode *pVnode);
+bool vnodeShouldCommit(SVnode *pVnode);
+int  vnodeAsyncCommit(SVnode *pVnode);
 
 #ifdef __cplusplus
 }

source/dnode/vnode/impl/inc/vnodeDef.h

@@ -16,10 +16,12 @@
 #ifndef _TD_VNODE_DEF_H_
 #define _TD_VNODE_DEF_H_
 
+#include "mallocator.h"
 #include "vnode.h"
 #include "vnodeAllocatorPool.h"
 #include "vnodeOptions.h"
 #include "vnodeStateMgr.h"
+#include "vnodeCommit.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -30,7 +32,7 @@ struct SVnode {
   SVnodeOptions   options;
   SVState         state;
   SVAllocatorPool pool;
-  SVMemAllocator* inuse;
+  SMemAllocator*  inuse;
   SMeta*          pMeta;
   STsdb*          pTsdb;
   STQ*            pTq;

source/dnode/vnode/impl/src/vnodeCommit.c

@@ -13,13 +13,15 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "vnodeInt.h"
+#include "vnodeDef.h"
 
 static int vnodeStartCommit(SVnode *pVnode);
 static int vnodeEndCommit(SVnode *pVnode);
 
+bool vnodeShouldCommit(SVnode *pVnode) { return false; }
+
 int vnodeAsyncCommit(SVnode *pVnode) {
-  #if 0
+#if 0
   if (vnodeStartCommit(pVnode) < 0) {
     // TODO
   }

source/dnode/vnode/impl/src/vnodeWrite.c

@@ -21,9 +21,39 @@ int vnodeProcessWriteReqs(SVnode *pVnode, SReqBatch *pReqBatch) {
 }
 
 int vnodeApplyWriteRequest(SVnode *pVnode, const SRequest *pRequest) {
-  int type;
+  int    reqType; /* TODO */
-  /* TODO */
+  size_t reqSize; /* TODO */
-  return 0;
+  int    code = 0;
+
+  // Copy the request to vnode buffer
+  SRequest *pReq = mMalloc(pVnode->inuse, reqSize);
+  if (pReq == NULL) {
+    // TODO: handle error
+  }
+
+  // Push the request to TQ so consumers can consume
+  tqPushMsg(pVnode->pTq, pReq, 0);
+
+  // Process the request
+  switch (reqType) {
+    case TSDB_MSG_TYPE_CREATE_TABLE:
+      code = metaCreateTable(pVnode->pMeta, NULL /* TODO */);
+      break;
+    case TSDB_MSG_TYPE_DROP_TABLE:
+      code = metaDropTable(pVnode->pMeta, 0 /* TODO */);
+      break;
+      /* TODO */
+    default:
+      break;
+  }
+
+  if (vnodeShouldCommit(pVnode)) {
+    if (vnodeAsyncCommit(pVnode) < 0) {
+      // TODO: handle error
+    }
+  }
+
+  return code;
 }
 
 /* ------------------------ STATIC METHODS ------------------------ */