josh/TCPlayerbotCore
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Core/DBLayer: Allow asynchronous queries with callbacks to be executed in prepared statement format as well.

Browse Source 
+ Cleanup and documentation

--HG--
branch : trunk
This commit is contained in:
Machiavelli
2010-12-23 16:05:43 +01:00
parent b6724bea03
commit 76e111fa07
5 changed files with 120 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/shared/Database/DatabaseWorkerPool.h
+90 -27
View File
@@ -51,6 +51,7 @@ template <class T>
class DatabaseWorkerPool
{
public:
/* Activity state */
DatabaseWorkerPool() :
m_queue(new ACE_Activation_Queue(new ACE_Message_Queue<ACE_MT_SYNCH>))
{
@@ -132,6 +133,11 @@ class DatabaseWorkerPool
sLog.outSQLDriver("All connections on databasepool %s closed.", m_connectionInfo.database.c_str());
}
/**
Delayed one-way statement methods.
*/
//! Enqueues a one-way SQL operation in string format that will be executed asynchronously.
void Execute(const char* sql)
{
if (!sql)
@@ -141,6 +147,7 @@ class DatabaseWorkerPool
Enqueue(task);
}
//! Enqueues a one-way SQL operation in string format -with variable args- that will be executed asynchronously.
void PExecute(const char* sql, ...)
{
if (!sql)
@@ -155,6 +162,18 @@ class DatabaseWorkerPool
Execute(szQuery);
}
//! Enqueues a one-way SQL operation in prepared statement format that will be executed asynchronously.
void Execute(PreparedStatement* stmt)
{
PreparedStatementTask* task = new PreparedStatementTask(stmt);
Enqueue(task);
}
/**
Direct syncrhonous one-way statement methods.
*/
//! Directly executes a one-way SQL operation in string format, that will block the calling thread until finished.
void DirectExecute(const char* sql)
{
if (!sql)
@@ -165,6 +184,7 @@ class DatabaseWorkerPool
t->Unlock();
}
//! Directly executes a one-way SQL operation in string format -with variable args-, that will block the calling thread until finished.
void DirectPExecute(const char* sql, ...)
{
if (!sql)
@@ -179,6 +199,20 @@ class DatabaseWorkerPool
return DirectExecute(szQuery);
}
//! Directly executes a one-way SQL operation in prepared statement format, that will block the calling thread until finished.
void DirectExecute(PreparedStatement* stmt)
{
T* t = GetFreeConnection();
t->Execute(stmt);
t->Unlock();
}
/**
Syncrhonous query (with resultset) methods.
*/
//! Directly executes an SQL query in string format that will block the calling thread until finished.
//! Returns reference counted auto pointer, no need for manual memory management in upper level code.
QueryResult Query(const char* sql, MySQLConnection* conn = NULL)
{
if (!conn)
@@ -193,6 +227,8 @@ class DatabaseWorkerPool
return QueryResult(result);
}
//! Directly executes an SQL query in string format -with variable args- that will block the calling thread until finished.
//! Returns reference counted auto pointer, no need for manual memory management in upper level code.
QueryResult PQuery(const char* sql, MySQLConnection* conn, ...)
{
if (!sql)
@@ -207,6 +243,8 @@ class DatabaseWorkerPool
return Query(szQuery, conn);
}
//! Directly executes an SQL query in string format -with variable args- that will block the calling thread until finished.
//! Returns reference counted auto pointer, no need for manual memory management in upper level code.
QueryResult PQuery(const char* sql, ...)
{
if (!sql)
@@ -221,7 +259,27 @@ class DatabaseWorkerPool
return Query(szQuery);
}
ACE_Future<QueryResult> AsyncQuery(const char* sql)
//! Directly executes an SQL query in prepared format that will block the calling thread until finished.
//! Returns reference counted auto pointer, no need for manual memory management in upper level code.
PreparedQueryResult Query(PreparedStatement* stmt)
{
T* t = GetFreeConnection();
PreparedResultSet* ret = t->Query(stmt);
t->Unlock();
if (!ret || !ret->GetRowCount())
return PreparedQueryResult(NULL);
return PreparedQueryResult(ret);
}
/**
Asynchronous query (with resultset) methods.
*/
//! Enqueues a query in string format that will set the value of the QueryResultFuture return object as soon as the query is executed.
//! The return value is then processed in ProcessQueryCallback methods.
QueryResultFuture AsyncQuery(const char* sql)
{
QueryResultFuture res;
BasicStatementTask* task = new BasicStatementTask(sql, res);
@@ -229,7 +287,9 @@ class DatabaseWorkerPool
return res; //! Actual return value has no use yet
}
ACE_Future<QueryResult> AsyncPQuery(const char* sql, ...)
//! Enqueues a query in string format -with variable args- that will set the value of the QueryResultFuture return object as soon as the query is executed.
//! The return value is then processed in ProcessQueryCallback methods.
QueryResultFuture AsyncPQuery(const char* sql, ...)
{
va_list ap;
char szQuery[MAX_QUERY_LEN];
@@ -240,6 +300,19 @@ class DatabaseWorkerPool
return AsyncQuery(szQuery);
}
//! Enqueues a query in prepared format that will set the value of the PreparedQueryResultFuture return object as soon as the query is executed.
//! The return value is then processed in ProcessQueryCallback methods.
PreparedQueryResultFuture AsyncQuery(PreparedStatement* stmt)
{
PreparedQueryResultFuture res;
PreparedStatementTask* task = new PreparedStatementTask(stmt, res);
Enqueue(task);
return res;
}
//! Enqueues a vector of SQL operations (can be both adhoc and prepared) that will set the value of the QueryResultHolderFuture
//! return object as soon as the query is executed.
//! The return value is then processed in ProcessQueryCallback methods.
QueryResultHolderFuture DelayQueryHolder(SQLQueryHolder* holder)
{
QueryResultHolderFuture res;
@@ -248,11 +321,18 @@ class DatabaseWorkerPool
return res; //! Fool compiler, has no use yet
}
/**
Transaction context methods.
*/
//! Begins an automanaged transaction pointer that will automatically rollback if not commited. (Autocommit=0)
SQLTransaction BeginTransaction()
{
return SQLTransaction(new Transaction);
}
//! Enqueues a collection of one-way SQL operations (can be both adhoc and prepared). The order in which these operations
//! were appended to the transaction will be respected during execution.
void CommitTransaction(SQLTransaction transaction)
{
if (sLog.GetSQLDriverQueryLogging())
@@ -293,24 +373,18 @@ class DatabaseWorkerPool
trans->Append(sql);
}
/**
Other
*/
//! Automanaged (internally) pointer to a prepared statement object for usage in upper level code.
//! This object is not tied to the prepared statement on the MySQL context yet until execution.
PreparedStatement* GetPreparedStatement(uint32 index)
{
return new PreparedStatement(index);
}
void Execute(PreparedStatement* stmt)
{
PreparedStatementTask* task = new PreparedStatementTask(stmt);
Enqueue(task);
}
void DirectExecute(PreparedStatement* stmt)
{
T* t = GetFreeConnection();
t->Execute(stmt);
t->Unlock();
}
//! Apply escape string'ing for current collation. (utf8)
void escape_string(std::string& str)
{
if (str.empty())
@@ -322,18 +396,7 @@ class DatabaseWorkerPool
delete[] buf;
}
PreparedQueryResult Query(PreparedStatement* stmt)
{
T* t = GetFreeConnection();
PreparedResultSet* ret = t->Query(stmt);
t->Unlock();
if (!ret || !ret->GetRowCount())
return PreparedQueryResult(NULL);
return PreparedQueryResult(ret);
}
//! Keeps all our MySQL connections alive, prevent the server from disconnecting us.
void KeepAlive()
{
/// Ping synchronous connections
src/server/shared/Database/PreparedStatement.cpp
+22 -1
View File
@@ -332,10 +332,19 @@ void MySQLPreparedStatement::setValue(MYSQL_BIND* param, enum_field_types type,
//- Execution
PreparedStatementTask::PreparedStatementTask(PreparedStatement* stmt) :
m_stmt(stmt)
m_stmt(stmt),
m_has_result(false)
{
}
PreparedStatementTask::PreparedStatementTask(PreparedStatement* stmt, PreparedQueryResultFuture result) :
m_stmt(stmt),
m_has_result(true),
m_result(result)
{
}
PreparedStatementTask::~PreparedStatementTask()
{
delete m_stmt;
@@ -343,5 +352,17 @@ PreparedStatementTask::~PreparedStatementTask()
bool PreparedStatementTask::Execute()
{
if (m_has_result)
{
PreparedResultSet* result = m_conn->Query(m_stmt);
if (!result || !result->GetRowCount())
{
m_result.set(PreparedQueryResult(NULL));
return false;
}
m_result.set(PreparedQueryResult(result));
return true;
}
return m_conn->Execute(m_stmt);
}
src/server/shared/Database/PreparedStatement.h
+6
View File
@@ -19,6 +19,7 @@
#define _PREPAREDSTATEMENT_H
#include "SQLOperation.h"
#include <ace/Future.h>
//- Union for data buffer (upper-level bind -> queue -> lower-level bind)
union PreparedStatementDataUnion
@@ -138,16 +139,21 @@ class MySQLPreparedStatement
MYSQL_BIND* m_bind;
};
typedef ACE_Future<PreparedQueryResult> PreparedQueryResultFuture;
//- Lower-level class, enqueuable operation
class PreparedStatementTask : public SQLOperation
{
public:
PreparedStatementTask(PreparedStatement* stmt);
PreparedStatementTask(PreparedStatement* stmt, PreparedQueryResultFuture result);
~PreparedStatementTask();
bool Execute();
protected:
PreparedStatement* m_stmt;
bool m_has_result;
PreparedQueryResultFuture m_result;
};
#endif
src/server/shared/Database/Transaction.h
+1 -1
View File
@@ -35,7 +35,7 @@ class Transaction
void Append(const char* sql);
void PAppend(const char* sql, ...);
size_t GetSize() { return m_queries.size(); }
size_t GetSize() const { return m_queries.size(); }
protected:
void Cleanup();
src/server/shared/Threading/Callback.h
+1
View File
@@ -23,6 +23,7 @@
#include "QueryResult.h"
typedef ACE_Future<QueryResult> QueryResultFuture;
typedef ACE_Future<PreparedQueryResult> PreparedQueryResultFuture;
/*! A simple template using ACE_Future to manage callbacks from the thread and object that
issued the request. <ParamType> is variable type of parameter that is used as parameter