DatabaseWorkerPool.h

Go to the documentation of this file.

/*
 * Copyright (C) 2008-2016 TrinityCore <http://www.trinitycore.org/>
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; either version 2 of the License, or (at your
 * option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#ifndef _DATABASEWORKERPOOL_H
#define _DATABASEWORKERPOOL_H

#include "Common.h"
#include "Callback.h"
#include "MySQLConnection.h"
#include "Transaction.h"
#include "DatabaseWorker.h"
#include "PreparedStatement.h"
#include "Log.h"
#include "QueryResult.h"
#include "QueryHolder.h"
#include "AdhocStatement.h"
#include "StringFormat.h"

#include <mysqld_error.h>
#include <memory>
#include <array>

class PingOperation : public SQLOperation
{
 bool Execute() override
 {
 m_conn->Ping();
 return true;
 }
};

template <class T>
class DatabaseWorkerPool
{
 private:
 enum InternalIndex
 {
 IDX_ASYNC,
 IDX_SYNCH,
 IDX_SIZE
 };

 public:
 /* Activity state */
 DatabaseWorkerPool();

 ~DatabaseWorkerPool()
 {
 _queue->Cancel();
 }

 void SetConnectionInfo(std::string const& infoString, uint8 const asyncThreads, uint8 const synchThreads);

 uint32 Open();

 void Close();

 bool PrepareStatements();

 inline MySQLConnectionInfo const* GetConnectionInfo() const
 {
 return _connectionInfo.get();
 }

 void Execute(const char* sql)
 {
 if (Trinity::IsFormatEmptyOrNull(sql))
 return;

 BasicStatementTask* task = new BasicStatementTask(sql);
 Enqueue(task);
 }

 template<typename Format, typename... Args>
 void PExecute(Format&& sql, Args&&... args)
 {
 if (Trinity::IsFormatEmptyOrNull(sql))
 return;

 Execute(Trinity::StringFormat(std::forward<Format>(sql), std::forward<Args>(args)...).c_str());
 }

 void Execute(PreparedStatement* stmt)
 {
 PreparedStatementTask* task = new PreparedStatementTask(stmt);
 Enqueue(task);
 }

 void DirectExecute(const char* sql)
 {
 if (!sql)
 return;

 T* connection = GetFreeConnection();
 connection->Execute(sql);
 connection->Unlock();
 }

 template<typename Format, typename... Args>
 void DirectPExecute(Format&& sql, Args&&... args)
 {
 if (Trinity::IsFormatEmptyOrNull(sql))
 return;

 DirectExecute(Trinity::StringFormat(std::forward<Format>(sql), std::forward<Args>(args)...).c_str());
 }

 void DirectExecute(PreparedStatement* stmt)
 {
 T* connection = GetFreeConnection();
 connection->Execute(stmt);
 connection->Unlock();

 delete stmt;
 }

 QueryResult Query(const char* sql, T* connection = nullptr);

 template<typename Format, typename... Args>
 QueryResult PQuery(Format&& sql, T* conn, Args&&... args)
 {
 if (Trinity::IsFormatEmptyOrNull(sql))
 return QueryResult(nullptr);

 return Query(Trinity::StringFormat(std::forward<Format>(sql), std::forward<Args>(args)...).c_str(), conn);
 }

 template<typename Format, typename... Args>
 QueryResult PQuery(Format&& sql, Args&&... args)
 {
 if (!sql)
 return QueryResult(nullptr);

 return Query(Trinity::StringFormat(std::forward<Format>(sql), std::forward<Args>(args)...).c_str());
 }

 PreparedQueryResult Query(PreparedStatement* stmt);

 QueryResultFuture AsyncQuery(const char* sql);

 template<typename Format, typename... Args>
 QueryResultFuture AsyncPQuery(Format&& sql, Args&&... args)
 {
 return AsyncQuery(Trinity::StringFormat(std::forward<Format>(sql), std::forward<Args>(args)...).c_str());
 }

 PreparedQueryResultFuture AsyncQuery(PreparedStatement* stmt);

 QueryResultHolderFuture DelayQueryHolder(SQLQueryHolder* holder);

 SQLTransaction BeginTransaction()
 {
 return SQLTransaction(new Transaction);
 }

 void CommitTransaction(SQLTransaction transaction);

 void DirectCommitTransaction(SQLTransaction& transaction);

 void ExecuteOrAppend(SQLTransaction& trans, PreparedStatement* stmt)
 {
 if (!trans)
 Execute(stmt);
 else
 trans->Append(stmt);
 }

 void ExecuteOrAppend(SQLTransaction& trans, const char* sql)
 {
 if (!trans)
 Execute(sql);
 else
 trans->Append(sql);
 }

 typedef typename T::Statements PreparedStatementIndex;

 PreparedStatement* GetPreparedStatement(PreparedStatementIndex index)
 {
 return new PreparedStatement(index);
 }

 void EscapeString(std::string& str);

 void KeepAlive();

 private:
 uint32 OpenConnections(InternalIndex type, uint8 numConnections);

 unsigned long EscapeString(char *to, const char *from, unsigned long length)
 {
 if (!to || !from || !length)
 return 0;

 return mysql_real_escape_string(
 _connections[IDX_SYNCH].front()->GetHandle(), to, from, length);
 }

 void Enqueue(SQLOperation* op)
 {
 _queue->Push(op);
 }

 T* GetFreeConnection();

 char const* GetDatabaseName() const
 {
 return _connectionInfo->database.c_str();
 }

 std::unique_ptr<ProducerConsumerQueue<SQLOperation*>> _queue;
 std::array<std::vector<std::unique_ptr<T>>, IDX_SIZE> _connections;
 std::unique_ptr<MySQLConnectionInfo> _connectionInfo;
 uint8 _async_threads, _synch_threads;
};

#endif