Fix stream_delete functionality. Now stream metadata, auto_incrementing_id and...

Fix stream_delete functionality. Now stream metadata, auto_incrementing_id and list of streams are cleaned up.

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)+"$")
 }
 

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;
 }

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;
 };