From 9e3209f55f644f6c13e0d6a2e4cf2173bbae6ee5 Mon Sep 17 00:00:00 2001
From: karnem <mikhail.karnevskiy@desy.de>
Date: Thu, 30 Nov 2023 15:43:08 +0100
Subject: [PATCH] Fix stream_delete functionality. Now stream metadata,
auto_incrementing_id and list of streams are cleaned up.
---
broker/src/asapo_broker/database/mongodb.go | 9 +++++++++
common/cpp/src/database/mongodb_client.cpp | 9 ++++++---
common/cpp/src/database/mongodb_client.h | 3 ++-
3 files changed, 17 insertions(+), 4 deletions(-)
diff --git a/broker/src/asapo_broker/database/mongodb.go b/broker/src/asapo_broker/database/mongodb.go
index fa2052b28..ac3f5c643 100644
--- a/broker/src/asapo_broker/database/mongodb.go
+++ b/broker/src/asapo_broker/database/mongodb.go
@@ -79,6 +79,7 @@ const pointer_collection_name = "current_location"
const pointer_field_name = "current_pointer"
const last_message_collection_name = "last_messages"
const last_message_field_name = "last_message"
+const auto_id_counter_name = "auto_id_counters"
const no_session_msg = "database client not created"
const already_connected_msg = "already connected"
@@ -1105,6 +1106,14 @@ func (db *Mongodb) deleteServiceMeta(request Request) error {
if err != nil {
return err
}
+ db.deleteDocumentsInCollection(request, auto_id_counter_name, "_id", ".*_"+escapeQuery(request.Stream)+"$")
+ if err != nil {
+ return err
+ }
+ db.deleteDocumentsInCollection(request, streams_info, "name", escapeQuery(request.Stream))
+ if err != nil {
+ return err
+ }
return db.deleteDocumentsInCollection(request, pointer_collection_name, "_id", ".*_"+escapeQuery(request.Stream)+"$")
}
diff --git a/common/cpp/src/database/mongodb_client.cpp b/common/cpp/src/database/mongodb_client.cpp
index 1f5b037f1..3723e1da5 100644
--- a/common/cpp/src/database/mongodb_client.cpp
+++ b/common/cpp/src/database/mongodb_client.cpp
@@ -861,11 +861,12 @@ Error MongoDBClient::DeleteCollection(const std::string& name) const {
}
Error MongoDBClient::DeleteDocumentsInCollection(const std::string& collection_name,
- const std::string& querystr) const {
+ const std::string& querystr,
+ const std::string& field_name="_id") const {
auto collection = mongoc_client_get_collection(client_, database_name_.c_str(), collection_name.c_str());
mongoc_collection_set_write_concern(collection, write_concern_);
bson_error_t error;
- auto query = BCON_NEW ("_id", BCON_REGEX(querystr.c_str(), "i"));
+ auto query = BCON_NEW (field_name.c_str(), BCON_REGEX(querystr.c_str(), "i"));
if (!mongoc_collection_delete_many(collection, query, NULL, NULL, &error)) {
return DBErrorTemplates::kDBError.Generate(error.message);
}
@@ -886,8 +887,10 @@ Error MongoDBClient::DeleteStream(const std::string& stream) const {
DeleteCollections(acks_col);
std::string querystr = ".*_" + EscapeQuery(stream_encoded) + "$";
DeleteDocumentsInCollection("current_location", querystr);
+ DeleteDocumentsInCollection("auto_id_counters", "^data_" + EscapeQuery(stream_encoded) + "$");
+ DeleteDocumentsInCollection("streams", "^" + EscapeQuery(stream_encoded) + "$", "name");
}
- DeleteDocumentsInCollection("meta", "^" + EscapeQuery(stream_encoded) + "$");
+ DeleteDocumentsInCollection("meta", "^st_" + EscapeQuery(stream_encoded) + "$");
ModifyPersistedStreamsList(stream, false);
return err;
}
diff --git a/common/cpp/src/database/mongodb_client.h b/common/cpp/src/database/mongodb_client.h
index e45d79521..1681d597a 100644
--- a/common/cpp/src/database/mongodb_client.h
+++ b/common/cpp/src/database/mongodb_client.h
@@ -89,7 +89,8 @@ class MongoDBClient final : public Database {
Error UpdateLastStreamInfo(const char* str, StreamInfo* info) const;
Error DeleteCollection(const std::string& name) const;
Error DeleteCollections(const std::string& prefix) const;
- Error DeleteDocumentsInCollection(const std::string& collection_name, const std::string& querystr) const;
+ Error DeleteDocumentsInCollection(const std::string& collection_name, const std::string& querystr,
+ const std::string& field_name) const;
Error InsertWithAutoId(const MessageMeta& file, uint64_t* id_inserted) const;
Error ModifyPersistedStreamsList(const std::string& stream_name, bool add) const;
};
--
GitLab