feat(Core/Database): port TrinityCore database API (#5611)

src/server/database/Database/DatabaseWorker.cpp

@@ -1,39 +1,46 @@
 /*
- * Copyright (C) 2016+     AzerothCore <www.azerothcore.org>
- * Copyright (C) 2008-2016 TrinityCore <http://www.trinitycore.org/>
- * Copyright (C) 2005-2009 MaNGOS <http://getmangos.com/>
+ * Copyright (C) 2016+ AzerothCore <www.azerothcore.org>, released under GNU GPL v2 license, you may redistribute it and/or modify it under version 2 of the License, or (at your option), any later version.
+ * Copyright (C) 2021+ WarheadCore <https://github.com/WarheadCore>
  */
 
 #include "DatabaseWorker.h"
+#include "ProducerConsumerQueue.h"
 #include "SQLOperation.h"
-#include "MySQLConnection.h"
-#include "MySQLThreading.h"
 
-DatabaseWorker::DatabaseWorker(ACE_Activation_Queue* new_queue, MySQLConnection* con) :
-    m_queue(new_queue),
-    m_conn(con)
+DatabaseWorker::DatabaseWorker(ProducerConsumerQueue<SQLOperation*>* newQueue, MySQLConnection* connection)
 {
-    /// Assign thread to task
-    activate();
+    _connection = connection;
+    _queue = newQueue;
+    _cancelationToken = false;
+    _workerThread = std::thread(&DatabaseWorker::WorkerThread, this);
 }
 
-int DatabaseWorker::svc()
+DatabaseWorker::~DatabaseWorker()
 {
-    if (!m_queue)
-        return -1;
+    _cancelationToken = true;
 
-    SQLOperation* request = nullptr;
-    while (1)
+    _queue->Cancel();
+
+    _workerThread.join();
+}
+
+void DatabaseWorker::WorkerThread()
+{
+    if (!_queue)
+        return;
+
+    for (;;)
     {
-        request = (SQLOperation*)(m_queue->dequeue());
-        if (!request)
-            break;
+        SQLOperation* operation = nullptr;
 
-        request->SetConnection(m_conn);
-        request->call();
+        _queue->WaitAndPop(operation);
 
-        delete request;
+        if (_cancelationToken || !operation)
+            return;
+
+        operation->SetConnection(_connection);
+        operation->call();
+
+        delete operation;
     }
-
-    return 0;
 }