Merge remote-tracking branch 'couchbase/unstable' into HEAD

http://ci-eventing.northscale.in/eventing-08.09.2019-20.29.pass.html Change-Id: I005dac0ffefa94440e412e5f6e244675580ba282
couchbase · Sep 9, 2019 · bd72552 · bd72552
2 parents 96f0479 + a24d88f
commit bd72552
Show file tree

Hide file tree

Showing 3 changed files with 87 additions and 149 deletions.
diff --git a/v8_consumer/CMakeLists.txt b/v8_consumer/CMakeLists.txt
@@ -18,6 +18,7 @@ INCLUDE (FindCouchbaseLibuv)
 INCLUDE (FindCouchbaseV8)
 INCLUDE (FindCouchbaseBreakpad)
 INCLUDE (../libs/query/FindEventingQuery.cmake)
+INCLUDE (FindCouchbaseNlohmannJson)
 
 SET(LIBS_HEADERS
         ../libs/include
@@ -52,7 +53,8 @@ INCLUDE_DIRECTORIES(AFTER
                      ${FLATBUFFERS_INCLUDE_DIR}
                      ${CMAKE_INSTALL_PREFIX}/include
                      ${CMAKE_CURRENT_SOURCE_DIR}/../third_party/inspector
-                     ${LIBS_HEADERS})
+                     ${LIBS_HEADERS}
+                     ${NLOHMANN_JSON_INCLUDE_DIR})
 
 IF (NOT WIN32)
   SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fno-omit-frame-pointer -DYY_NEVER_INTERACTIVE -DYY_NO_UNPUT -DYY_NO_INPUT")

diff --git a/v8_consumer/include/v8worker.h b/v8_consumer/include/v8worker.h
@@ -25,6 +25,7 @@
 #include <libplatform/libplatform.h>
 #include <list>
 #include <map>
+#include <memory>
 #include <regex>
 #include <sstream>
 #include <string>
@@ -167,7 +168,8 @@ enum RETURN_CODE {
   kFailedInitBucketHandle,
   kOnUpdateCallFail,
   kOnDeleteCallFail,
-  kToLocalFailed
+  kToLocalFailed,
+  kJSONParseFailed
 };
 
 class Bucket;
@@ -216,11 +218,12 @@ class V8Worker {
 
   int V8WorkerLoad(std::string source_s);
   void RouteMessage();
+  bool IsValidDCPEvent(const std::unique_ptr<WorkerMessage> &msg,
+                       dcp_opcode event_type);
   void TaskDurationWatcher();
 
-  int SendUpdate(std::string value, std::string meta, int vb_no,
-                 uint64_t seq_no, std::string doc_type);
-  int SendDelete(std::string meta, int vb_no, uint64_t seq_no);
+  int SendUpdate(const std::string &value, const std::string &meta);
+  int SendDelete(const std::string &meta);
   void SendTimer(std::string callback, std::string timer_ctx);
   std::string CompileHandler(std::string handler);
   CodeVersion IdentifyVersion(std::string handler);
@@ -258,9 +261,11 @@ class V8Worker {
 
   void FilterUnlock();
 
-  int ParseMetadata(const std::string &metadata, int &vb_no, uint64_t &seq_no);
-  int ParseMetadataWithAck(const std::string &metadata, int &vb_no,
-                           uint64_t &seq_no, int &skip_ack, bool ack_check);
+  int ParseMetadata(const std::string &metadata, int &vb_no,
+                    uint64_t &seq_no) const;
+  int ParseMetadataWithAck(const std::string &metadata_str, int &vb_no,
+                           uint64_t &seq_no, int &skip_ack,
+                           bool ack_check) const;
 
   void SetThreadExitFlag();