CXXCBC-173: complete streaming lexer even if pointer didn't match

Author: avsej

couchbase/utils/json_streaming_lexer.cxx

@@ -310,7 +310,7 @@ row_pop_callback(jsonsl_t lexer, jsonsl_action_t /* action */, struct jsonsl_sta
 }
 
 static void
-initial_action_pop_callback(jsonsl_t lexer, jsonsl_action_t /* action */, struct jsonsl_state_st* state, const jsonsl_char_t* /* at */)
+initial_action_pop_callback(jsonsl_t lexer, jsonsl_action_t action, struct jsonsl_state_st* state, const jsonsl_char_t* at)
 {
     auto* impl = static_cast<detail::streaming_lexer_impl*>(lexer->data);
 
@@ -321,6 +321,10 @@ initial_action_pop_callback(jsonsl_t lexer, jsonsl_action_t /* action */, struct
     if (state->type == JSONSL_T_HKEY) {
         impl->last_key_ = impl->buffer_.substr(state->pos_begin + 1, state->pos_cur - state->pos_begin - 1);
     }
+
+    if (state->data == STATE_MARKER_ROOT) {
+        trailer_pop_callback(lexer, action, state, at);
+    }
 }
 
 static void

test/test_integration_query.cxx

@@ -405,3 +405,25 @@ TEST_CASE("integration: sticking query to the service node", "[integration]")
         REQUIRE(std::set(used_nodes.begin(), used_nodes.end()).size() == 1);
     }
 }
+
+TEST_CASE("analytics create dataset")
+{
+    test::utils::integration_test_guard integration;
+
+    if (!integration.cluster_version().supports_analytics() || !integration.has_analytics_service() ||
+        !integration.cluster_version().supports_collections()) {
+        return;
+    }
+    if (!integration.cluster_version().supports_gcccp()) {
+        test::utils::open_bucket(integration.cluster, integration.ctx.bucket);
+    }
+    couchbase::operations::analytics_request req{ fmt::format("CREATE DATAVERSE `{}`.`test-scope` IF NOT EXISTS", integration.ctx.bucket) };
+    std::vector<std::string> rows{};
+    req.row_callback = [&rows](std::string&& row) {
+        rows.emplace_back(std::move(row));
+        return couchbase::utils::json::stream_control::next_row;
+    };
+
+    auto resp = test::utils::execute(integration.cluster, req);
+    REQUIRE_FALSE(resp.ctx.ec);
+}

test/test_unit_json_streaming_lexer.cxx

@@ -185,3 +185,50 @@ TEST_CASE("unit: json_streaming_lexer parse chunked metadata trailer", "[unit]")
     REQUIRE(result.meta == expected_meta);
     REQUIRE(result.rows[0] == R"(42)");
 }
+
+TEST_CASE("unit: json_streaming_lexer parse payload with missing results", "[unit]")
+{
+    test::utils::init_logger();
+
+    std::string chunk = R"(
+{
+
+	"requestID": "d07c0cde-cd80-4620-bb6b-d0641f272420",
+	"clientContextID": "a7bbe750-20a2-4e46-eb67-315e3733b2a8",
+	"signature": {
+		"*": "*"
+	},
+	"plans":{},
+	"status": "success",
+	"metrics": {
+		"elapsedTime": "6.56579ms",
+		"executionTime": "5.552905ms",
+		"resultCount": 0,
+		"resultSize": 0,
+		"processedObjects": 0
+	}
+}
+)";
+    couchbase::utils::json::streaming_lexer lexer("/results/^", 4);
+    query_result result{};
+    bool on_row_handler_executed = false;
+    lexer.on_row([&result, &on_row_handler_executed](std::string&& row) {
+        on_row_handler_executed = true;
+        result.rows.emplace_back(std::move(row));
+        return couchbase::utils::json::stream_control::next_row;
+    });
+    bool on_complete_handler_excecuted = false;
+    lexer.on_complete([&result, &on_complete_handler_excecuted](std::error_code ec, std::size_t number_of_rows, std::string&& meta) {
+        on_complete_handler_excecuted = true;
+        result.ec = ec;
+        result.number_of_rows = number_of_rows;
+        result.meta = std::move(meta);
+    });
+    lexer.feed(chunk);
+    REQUIRE_FALSE(on_row_handler_executed);
+    REQUIRE(on_complete_handler_excecuted);
+    REQUIRE_FALSE(result.ec);
+    REQUIRE(result.number_of_rows == 0);
+    REQUIRE(result.rows.empty());
+    REQUIRE(result.meta == chunk);
+}