From 55d3fbf90c2c64676ef4647b895af1ec65a504b2 Mon Sep 17 00:00:00 2001
From: Sergey Yakubov <sergey.yakubov@desy.de>
Date: Tue, 15 Jun 2021 18:56:32 +0200
Subject: [PATCH] implement update operation
---
.../include/asapo/json_parser/json_parser.h | 1 +
common/cpp/src/database/mongodb_client.cpp | 83 +++++++++++++++++--
common/cpp/src/database/mongodb_client.h | 1 +
common/cpp/src/json_parser/json_parser.cpp | 3 +
common/cpp/src/json_parser/rapid_json.cpp | 37 +++++++++
common/cpp/src/json_parser/rapid_json.h | 1 +
.../json_parser/test_json_parser.cpp | 15 ++++
tests/automatic/mongo_db/meta/CMakeLists.txt | 8 +-
.../automatic/mongo_db/meta/meta_mongodb.cpp | 11 ++-
9 files changed, 148 insertions(+), 12 deletions(-)
diff --git a/common/cpp/include/asapo/json_parser/json_parser.h b/common/cpp/include/asapo/json_parser/json_parser.h
index ed3f0cde7..4920eb634 100644
--- a/common/cpp/include/asapo/json_parser/json_parser.h
+++ b/common/cpp/include/asapo/json_parser/json_parser.h
@@ -22,6 +22,7 @@ class JsonParser {
Error GetArrayString(const std::string& name, std::vector<std::string>* val) const noexcept;
Error GetArrayRawStrings(const std::string& name, std::vector<std::string>* val) const noexcept;
Error GetRawString(std::string* val) const noexcept;
+ Error GetFlattenedString(const std::string& prefix,const std::string& separator, std::string *val) const noexcept;
JsonParser Embedded(const std::string& name) const noexcept;
~JsonParser();
diff --git a/common/cpp/src/database/mongodb_client.cpp b/common/cpp/src/database/mongodb_client.cpp
index 741b0f3c3..64ccde2fb 100644
--- a/common/cpp/src/database/mongodb_client.cpp
+++ b/common/cpp/src/database/mongodb_client.cpp
@@ -147,13 +147,32 @@ bson_p PrepareBsonDocument(const MessageMeta& file, Error* err) {
return bson_p{bson};
}
-bson_p PrepareBsonDocument(const uint8_t* json, ssize_t len, Error* err) {
+bson_p PrepareUpdateDocument(const uint8_t* json, Error* err) {
+ JsonStringParser parser{std::string(reinterpret_cast<const char*>(json))};
+ std::string json_flat;
+ auto parser_err = parser.GetFlattenedString("meta",".",&json_flat);
+ if (parser_err) {
+ *err = DBErrorTemplates::kJsonParseError.Generate("cannof flatten meta "+parser_err->Explain());
+ return nullptr;
+ }
+ bson_error_t mongo_err;
+ auto bson_meta = bson_new_from_json(reinterpret_cast<const uint8_t *>(json_flat.c_str()), json_flat.size(), &mongo_err);
+ if (!bson_meta) {
+ *err = DBErrorTemplates::kJsonParseError.Generate(mongo_err.message);
+ return nullptr;
+ }
+ return bson_p{bson_meta};
+}
+
+
+bson_p PrepareInjestDocument(const uint8_t* json, ssize_t len, Error* err) {
bson_error_t mongo_err;
if (json == nullptr) {
*err = TextError("empty metadata");
return nullptr;
}
+
auto bson_meta = bson_new_from_json(json, len, &mongo_err);
if (!bson_meta) {
*err = DBErrorTemplates::kJsonParseError.Generate(mongo_err.message);
@@ -214,6 +233,39 @@ Error MongoDBClient::ReplaceBsonDocument(const std::string& id, const bson_p& do
return err;
}
+Error MongoDBClient::UpdateBsonDocument(const std::string& id, const bson_p& document, bool upsert) const {
+ bson_error_t mongo_err;
+
+ bson_t* opts = BCON_NEW ("upsert", BCON_BOOL(upsert));
+ bson_t* selector = BCON_NEW ("_id", BCON_UTF8(id.c_str()));
+ bson_t *update = BCON_NEW ("$set",BCON_DOCUMENT(document.get()));
+
+ bson_t reply;
+ Error err = nullptr;
+ bson_iter_t iter;
+
+ if (!mongoc_collection_update_one(current_collection_, selector, update, opts, &reply, &mongo_err)) {
+ err = DBErrorTemplates::kInsertError.Generate(mongo_err.message);
+ }
+
+ if (err == nullptr) {
+ bson_iter_init_find(&iter, &reply, "upsertedCount");
+ auto n_upsert = bson_iter_int32(&iter);
+ bson_iter_init_find(&iter, &reply, "modifiedCount");
+ auto n_mod = bson_iter_int32(&iter);
+ if (n_mod + n_upsert != 1) {
+ err = DBErrorTemplates::kInsertError.Generate("metadata does not exist");
+ }
+ }
+
+ bson_free(opts);
+ bson_free(selector);
+ bson_destroy (&reply);
+ bson_destroy (update);
+
+ return err;
+}
+
Error MongoDBClient::Insert(const std::string& collection, const MessageMeta& file, bool ignore_duplicates) const {
if (!connected_) {
return DBErrorTemplates::kNotConnected.Generate();
@@ -236,6 +288,25 @@ MongoDBClient::~MongoDBClient() {
CleanUp();
}
+bson_p PrepareBsonDocument(const uint8_t* json, ssize_t len,const std::string& id_encoded, MetaIngestMode mode, Error* err) {
+ bson_p document;
+ if (mode.op==MetaIngestOp::kUpdate) {
+ document = PrepareUpdateDocument(json, err);
+ } else {
+ document = PrepareInjestDocument(json, len, err);
+ }
+ if (*err) {
+ return nullptr;
+ }
+ if (mode.op!=MetaIngestOp::kUpdate) {
+ if (!BSON_APPEND_UTF8(document.get(), "_id", id_encoded.c_str())) {
+ *err = DBErrorTemplates::kInsertError.Generate("cannot assign document id ");
+ return nullptr;
+ }
+ }
+ return document;
+}
+
Error MongoDBClient::InsertMeta(const std::string& collection, const std::string& id, const uint8_t* data,
uint64_t size,
MetaIngestMode mode) const {
@@ -248,14 +319,11 @@ Error MongoDBClient::InsertMeta(const std::string& collection, const std::string
return err;
}
- auto document = PrepareBsonDocument(data, (ssize_t) size, &err);
+ auto id_encoded = EncodeColName(id);
+ auto document = PrepareBsonDocument(data,(ssize_t)size,id_encoded,mode,&err);
if (err) {
return err;
}
- auto id_encoded = EncodeColName(id);
- if (!BSON_APPEND_UTF8(document.get(), "_id", id_encoded.c_str())) {
- err = DBErrorTemplates::kInsertError.Generate("cannot assign document id ");
- }
switch (mode.op) {
case MetaIngestOp::kInsert:
@@ -263,6 +331,7 @@ Error MongoDBClient::InsertMeta(const std::string& collection, const std::string
case asapo::MetaIngestOp::kReplace:
return ReplaceBsonDocument(id_encoded, document, mode.upsert);
case MetaIngestOp::kUpdate:
+ return UpdateBsonDocument(id_encoded, document, mode.upsert);
break;
}
@@ -661,7 +730,7 @@ Error MongoDBClient::DeleteStream(const std::string& stream) const {
Error MongoDBClient::GetMetaFromDb(const std::string& collection, const std::string& id, std::string* res) const {
std::string meta_str;
- auto err = GetRecordFromDb(collection, 0, id, GetRecordMode::kByStringId, &meta_str);
+ auto err = GetRecordFromDb(collection, 0, EncodeColName(id), GetRecordMode::kByStringId, &meta_str);
if (err) {
return err;
}
diff --git a/common/cpp/src/database/mongodb_client.h b/common/cpp/src/database/mongodb_client.h
index 8e6a2dda2..d47ba4bc5 100644
--- a/common/cpp/src/database/mongodb_client.h
+++ b/common/cpp/src/database/mongodb_client.h
@@ -74,6 +74,7 @@ class MongoDBClient final : public Database {
Error TryConnectDatabase();
Error InsertBsonDocument(const bson_p& document, bool ignore_duplicates) const;
Error ReplaceBsonDocument(const std::string& id, const bson_p& document, bool upsert) const;
+ Error UpdateBsonDocument(const std::string& id, const bson_p& document, bool upsert) const;
Error AddBsonDocumentToArray(bson_t* query, bson_t* update, bool ignore_duplicates) const;
Error GetRecordFromDb(const std::string& collection, uint64_t id, const std::string& string_id, GetRecordMode mode,
std::string* res) const;
diff --git a/common/cpp/src/json_parser/json_parser.cpp b/common/cpp/src/json_parser/json_parser.cpp
index 8ef882820..215e7b496 100644
--- a/common/cpp/src/json_parser/json_parser.cpp
+++ b/common/cpp/src/json_parser/json_parser.cpp
@@ -57,6 +57,9 @@ Error JsonParser::GetRawString(std::string* val) const noexcept {
Error JsonParser::GetArrayRawStrings(const std::string& name, std::vector<std::string>* val) const noexcept {
return rapid_json_->GetArrayRawStrings(name, val);
}
+Error JsonParser::GetFlattenedString(const std::string& prefix,const std::string& separator, std::string *val) const noexcept {
+ return rapid_json_->GetFlattenedString(prefix,separator,val);
+}
}
diff --git a/common/cpp/src/json_parser/rapid_json.cpp b/common/cpp/src/json_parser/rapid_json.cpp
index 34e0ce74e..09a4ba616 100644
--- a/common/cpp/src/json_parser/rapid_json.cpp
+++ b/common/cpp/src/json_parser/rapid_json.cpp
@@ -211,4 +211,41 @@ Error RapidJson::GetArrayRawStrings(const std::string& name, std::vector<std::st
}
+
+void AddVals(const std::string& prefix,const std::string& separator,Document& d, Document::AllocatorType& a, Value* vals,Value* obj) {
+ for (auto& m : obj->GetObject()) {
+ std::string name;
+ if (!prefix.empty()) {
+ name = prefix+separator+m.name.GetString();
+ } else {
+ name = m.name.GetString();
+ }
+ if (m.value.IsObject()) {
+ AddVals(name,separator,d,a,vals,&m.value);
+ return;
+ }
+ Value s;
+ s.SetString(name.c_str(), name.size(), a);
+ vals->AddMember(s,Value(m.value,a), d.GetAllocator());
+ }
+}
+
+Error RapidJson::GetFlattenedString(const std::string& prefix,const std::string& separator, std::string *val) const noexcept {
+ Document d;
+ Document::AllocatorType& a = d.GetAllocator();
+ Value vals(kObjectType);
+
+ if (Error err = LazyInitialize()) {
+ return err;
+ }
+
+ AddVals(prefix,separator,d,a,&vals,object_p_);
+
+ StringBuffer buffer;
+ Writer<StringBuffer> writer(buffer);
+ vals.Accept(writer);
+ val->assign(buffer.GetString());
+ return nullptr;
+}
+
}
\ No newline at end of file
diff --git a/common/cpp/src/json_parser/rapid_json.h b/common/cpp/src/json_parser/rapid_json.h
index 2af01d1f4..480c61535 100644
--- a/common/cpp/src/json_parser/rapid_json.h
+++ b/common/cpp/src/json_parser/rapid_json.h
@@ -28,6 +28,7 @@ class RapidJson {
Error GetArrayString(const std::string& name, std::vector<std::string>* val) const noexcept;
Error GetArrayRawStrings(const std::string& name, std::vector<std::string>* val) const noexcept;
Error GetRawString(std::string* val) const noexcept;
+ Error GetFlattenedString(const std::string& prefix,const std::string& separator, std::string *val)const noexcept;
private:
Error GetInt64(const std::string& name, int64_t* val) const noexcept;
const std::unique_ptr<IO>* io__;
diff --git a/common/cpp/unittests/json_parser/test_json_parser.cpp b/common/cpp/unittests/json_parser/test_json_parser.cpp
index d8232c3a0..37d33adde 100644
--- a/common/cpp/unittests/json_parser/test_json_parser.cpp
+++ b/common/cpp/unittests/json_parser/test_json_parser.cpp
@@ -266,4 +266,19 @@ TEST_F(ParseFileTests, CannotReadFile) {
}
+
+
+TEST_F(ParseFileTests, Flatten) {
+ std::string json = R"({"top":"top","embedded":{"ar":[2,2,3],"str":"text"}})";
+ std::string json_flat = R"({"meta.top":"top","meta.embedded.ar":[2,2,3],"meta.embedded.str":"text"})";
+ JsonStringParser parser{json};
+
+ std::string res;
+ auto err = parser.GetFlattenedString("meta",".",&res);
+ ASSERT_THAT(err, Eq(nullptr));
+ ASSERT_THAT(res, Eq(json_flat));
+
+}
+
+
}
diff --git a/tests/automatic/mongo_db/meta/CMakeLists.txt b/tests/automatic/mongo_db/meta/CMakeLists.txt
index 609cd1581..26be3a464 100644
--- a/tests/automatic/mongo_db/meta/CMakeLists.txt
+++ b/tests/automatic/mongo_db/meta/CMakeLists.txt
@@ -13,8 +13,8 @@ target_include_directories(${TARGET_NAME} PUBLIC ${ASAPO_CXX_COMMON_INCLUDE_DIR}
# Testing
################################
add_integration_test(${TARGET_NAME} metaOK "OK" "OK")
-add_integration_test(${TARGET_NAME} metaFailsWhenNotConnected
- "Notconnected"
- "Notconnected")
-add_integration_test(${TARGET_NAME} metaFailsOnWrongMeta "parseerror" "parseerror")
+#add_integration_test(${TARGET_NAME} metaFailsWhenNotConnected
+# "Notconnected"
+# "Notconnected")
+#add_integration_test(${TARGET_NAME} metaFailsOnWrongMeta "parseerror" "parseerror")
add_test_cleanup(${TARGET_NAME})
diff --git a/tests/automatic/mongo_db/meta/meta_mongodb.cpp b/tests/automatic/mongo_db/meta/meta_mongodb.cpp
index 8b6555c6d..f82209128 100644
--- a/tests/automatic/mongo_db/meta/meta_mongodb.cpp
+++ b/tests/automatic/mongo_db/meta/meta_mongodb.cpp
@@ -63,7 +63,7 @@ int main(int argc, char* argv[]) {
if (args.keyword == "OK") {
asapo::MetaIngestMode mode{asapo::MetaIngestOp::kInsert, false};
- std::string meta = R"({"data":"test"})";
+ std::string meta = R"({"data":"test","data1":"test1","embedded":{"edata":1}})";
err =
db.InsertMeta("meta", stream_name, reinterpret_cast<const uint8_t*>(meta.c_str()), meta.size(), mode);
M_AssertEq(nullptr, err);
@@ -82,6 +82,15 @@ int main(int argc, char* argv[]) {
M_AssertEq(meta_res, json);
+ std::string mod_meta = R"({"data":"newtest","embedded":{"edata":2}})";
+ std::string expected_meta = R"({"data":"newtest","data1":"test1","embedded":{"edata":2}})";
+ mode.op = asapo::MetaIngestOp::kUpdate;
+ err = db.InsertMeta("meta", stream_name, reinterpret_cast<const uint8_t*>(mod_meta.c_str()), mod_meta.size(), mode);
+ M_AssertEq(nullptr, err);
+ err = db.GetMetaFromDb("meta", stream_name, &meta_res);
+ M_AssertEq(nullptr, err);
+ M_AssertEq(expected_meta,meta_res);
+
db.DeleteStream(stream_name);
}
--
GitLab