Event Driven Database

void *          m_pNotEmptyEvent;

#include "platform/platform.h"
#include "platform/platformMutex.h"
#include "platform/platformSemaphore.h"
#include "platform/gameInterface.h"

DatabaseInterface::DatabaseInterface()
:  m_pEventQueue(&m_EventQueue1)
,  m_pEventQueueMutex(Mutex::createMutex())
,  m_pNotEmptyEvent(Semaphore::createSemaphore(0))
,  m_ShuttingDown(false)
,  m_con(false)
{
   m_ProcessThread = new Thread((ThreadRunFunction)processThread, (S32) this, 1);
}

void DatabaseInterface::postEvent(DatabaseRequest& event)
{
   // Only one thread can post at a time.
   Mutex::lockMutex(m_pEventQueueMutex);

   // Create a deep copy of event, and save a pointer to the copy in a vector.
   DatabaseRequest* copy = (DatabaseRequest*)dMalloc(event.size);
   dMemcpy(copy, &event, event.size);
   m_pEventQueue->push_back(copy);

   // Assert the condition variable
   Semaphore::releaseSemaphore(m_pNotEmptyEvent);

   Mutex::unlockMutex(m_pEventQueueMutex);

}

void DatabaseInterface::processLoop()
{
   if (!m_con.connect(
      Con::getVariable("Database::database"), 
      Con::getVariable("Database::host"), 
      Con::getVariable("Database::user"), 
      Con::getVariable("Database::password")))
   {
      Con::printf("Error connecting to database!");

      // TODO: Loop a few times before giving up?
      return;
   }

   while(!m_ShuttingDown)
   {
      // Wait for the event to be signalled
      Semaphore::acquireSemaphore(m_pNotEmptyEvent, true);

      Mutex::lockMutex(m_pEventQueueMutex);

      // swap event queue pointers
      Vector<DatabaseRequest*> &fullEventQueue = *m_pEventQueue;
      if(m_pEventQueue == &m_EventQueue1)
         m_pEventQueue = &m_EventQueue2;
      else
         m_pEventQueue = &m_EventQueue1;

      Mutex::unlockMutex(m_pEventQueueMutex);

      // Keep track of the original size
      const int size = fullEventQueue.size();

      // Walk the event queue in fifo order, processing the events, then clear the queue.
      for(int i=0; i < size; i++)
      {
         processEvent(fullEventQueue[i]);
         dFree(fullEventQueue[i]);
      }
      fullEventQueue.clear();
   }
}

//-----------------------------------------------------------------------------
// Copyright (C) 2007 Tony "Sgt Flame" Richards
// 
// Feel free to use this in your TGE / TGEA games, but if you use it
// then please mention my name in your "credits" section of your game.
// If you have any questions or problems, feel free to contact me:
//
// E-mail:  SgtFlame@GamesBySarge.com
//
//-----------------------------------------------------------------------------
#ifdef DEDICATED

#ifndef _RECORDSET_H_
#define _RECORDSET_H_

#ifndef _SIMBASE_H_
#include "console/simBase.h"
#endif

#include <vector>
#include <map>
#include <string>
#include <list>

class RecordSet 
: public SimObject
{
   //-~-~-~-~-~
   // Typedefs
private:
   typedef SimObject Parent;

   //-~-~-~-~-~-~
   // 'Structors
public:
    RecordSet();
   ~RecordSet();

   //-~-~-~-~-~-~-~-~-~-~-~-~-~
   // SimObject implementation
public:
   virtual bool onAdd();
   virtual void onRemove();

   static void initPersistFields();

   //-~-~-~-~-~-~-~-~-~-~-~-~-~
   // Getter / Setter methods
public:
   void setQueryId(S32 queryId);
   S32 getQueryId();

   void setFieldCount(S32 fieldCount);
   S32 getFieldCount();

   void setFieldName(S32 index, const char* pFieldName);
   const char* getFieldName(S32 index);

   void setFieldValue(S32 index, const char* pValue);
   const char* getFieldValue(S32 index);
   const char* getFieldValue(const char* pFieldName);

   //-~-~-~-~-~-~-~-~-~
   // Record traversal
public:
   void addRecord();
   bool nextRecord();

   //-~-~-~-~-~-~-~-~-~
   // Member variables
private:
   typedef std::vector<std::string> record_type;
   typedef std::list<record_type>   record_set_type;

   S32 m_queryId;
   S32 m_fieldCount;

   /// Field names kept in order
   std::vector<std::string>   m_fieldNames;

   /// Maps the name of the field to the index
   /// This is slightly inefficient to store the
   /// full copy of the string instead of a reference
   /// since it's already being stored in m_fieldNames,
   /// but I'm lazy and it's really not an important
   /// optimization unless you have a ton of recordsets
   /// hanging around.
   std::map<std::string, S32> m_fieldNameMap;

   /// Record set
   record_set_type            m_recordSet;

   bool                       m_firstRecord;
   record_set_type::iterator  m_currentRecord;

   //-~-~-~-~-~-~-~-~-~-~-~
   // Console Object Stuff
public:
   DECLARE_CONOBJECT(RecordSet);

}; // class RecordSet

#endif // _RECORDSET_H_

#endif // DEDICATED

//-----------------------------------------------------------------------------
// Copyright (C) 2007 Tony "Sgt Flame" Richards
// 
// Feel free to use this in your TGE / TGEA games, but if you use it
// then please mention my name in your "credits" section of your game.
// If you have any questions or problems, feel free to contact me:
//
// E-mail:  SgtFlame@GamesBySarge.com
//
//-----------------------------------------------------------------------------
#ifdef DEDICATED

#include "console/recordSet.h"

#include "console/simBase.h"
#include "console/consoleInternal.h"

#include "core/databaseInterface.h"

#include <algorithm>
#include <cctype>

IMPLEMENT_CONOBJECT(RecordSet);

RecordSet::RecordSet()
:  m_queryId(0)
,  m_fieldCount(0)
,  m_firstRecord(true)
{
}

RecordSet::~RecordSet()
{
}

bool RecordSet::onAdd()
{
   if (!Parent::onAdd())
      return false;

   const char *name = getName();
   if(name && name[0] && getClassRep())
   {
      Namespace *parent = getClassRep()->getNameSpace();
      Con::linkNamespaces(parent->mName, name);
      mNameSpace = Con::lookupNamespace(name);
   }

   return true;
}

void RecordSet::onRemove()
{
   Parent::onRemove();
}

void RecordSet::initPersistFields()
{
   Parent::initPersistFields();
}

void RecordSet::setQueryId(S32 queryId)
{
   m_queryId = queryId;
}

S32 RecordSet::getQueryId()
{
   return m_queryId;
}

void RecordSet::setFieldCount(S32 fieldCount)
{
   // We don't really need to keep track of this, but we do use it
   // as a hint for the vector size.
   m_fieldCount = fieldCount;
   m_fieldNames.resize(fieldCount);
}

S32 RecordSet::getFieldCount()
{
   return m_fieldCount;
}

void RecordSet::setFieldName(S32 index, const char* pFieldName)
{
   std::string tmpFieldName(pFieldName);
   m_fieldNames[index] = tmpFieldName;
   m_fieldNameMap[tmpFieldName] = index;
}

const char* RecordSet::getFieldName(S32 index)
{
   return m_fieldNames[index].c_str();
}

void RecordSet::setFieldValue(S32 index, const char *pValue)
{
   m_recordSet.back()[index] = pValue;
}

const char* RecordSet::getFieldValue(S32 index)
{
   if (m_firstRecord)
   {
      Con::errorf("Call RecordSet::nextRecord() before calling getFieldValue");
      return "";
   }

   return (*m_currentRecord)[index].c_str();
}

const char* RecordSet::getFieldValue(const char* pFieldName)
{
   std::string tmpFieldName(pFieldName);
   std::transform(tmpFieldName.begin(), tmpFieldName.end(), tmpFieldName.begin(), 
      (int(*)(int)) std::tolower);

   // TODO: Use find() to see if the field name actually 
   // exists and display an error if it doesn't exist.
   // For now, this will return the value of the first field
   // if the specified field name doesn't exist.
   return getFieldValue(m_fieldNameMap[tmpFieldName]);
}

void RecordSet::addRecord()
{
   record_type tmpRecord;
   m_recordSet.push_back(tmpRecord);
   m_recordSet.back().resize(m_fieldCount);
}

bool RecordSet::nextRecord()
{
   if (m_firstRecord)
   {
      m_firstRecord = false;
      m_currentRecord = m_recordSet.begin();
   }
   else
   {
      ++m_currentRecord;
   }

   return m_currentRecord != m_recordSet.end();
}

ConsoleMethod(RecordSet, getQueryId, S32, 2, 2, "getQueryId()")
{
   return object->getQueryId();
}

ConsoleMethod(RecordSet, nextRecord, bool, 2, 2, "nextRecord()")
{
   return object->nextRecord();
}

ConsoleMethod(RecordSet, getFieldCount, S32, 2, 2, "getFieldCount()")
{
   return object->getFieldCount();
}

ConsoleMethod(RecordSet, getFieldName, const char*, 3, 3, "getFieldName(%index)")
{
   return object->getFieldName(dAtoi(argv[2]));
}

ConsoleMethod(RecordSet, getFieldValue, const char*, 3, 3, "getFieldValue(%index or %fieldName)")
{
   if (dIsalpha(argv[2][0]))
   {
      return object->getFieldValue(argv[2]);
   }
   else
   {
      return object->getFieldValue(dAtoi(argv[2]));
   }
}

static int gStatementHandle = 0;

ConsoleFunction(query, S32, 3, 3, "query(%sqlStatement, %callback)")
{
   GenericQueryRequest request;
   request.pRecordSet = new RecordSet;
   request.pRecordSet->registerObject();
   request.pRecordSet->setQueryId(gStatementHandle++);

   request.pSQLString = dStrdup(argv[1]);
   request.pQueryCallback = dStrdup(argv[2]);

   gDatabase->postEvent(request);

   return request.pRecordSet->getQueryId();
}

#endif // DEDICATED

enum DatabaseEventType
{
... possibly more types ...
   GenericQuery
};

class RecordSet;

struct GenericQueryRequest
: public DatabaseRequest
{
   GenericQueryRequest() { size = sizeof(GenericQueryRequest); type2 = GenericQuery; }

   const char*    pSQLString;
   const char*    pQueryCallback;
   RecordSet*     pRecordSet;
};

struct GenericQueryReply
: public DatabaseReply
{
   GenericQueryReply() { size = sizeof(GenericQueryReply); type2 = GenericQuery; }

   const char*    pQueryCallback;
   RecordSet*     pRecordSet;
};

void DatabaseInterface::processEvent(DatabaseRequest* event)
{
   try
   {
      // First, always ping to attempt to reconnect if we've dropped
      if (!m_con.connected())
      {
         Con::printf("Lost database connection... reconnecting");

         if (!m_con.connect(
            Con::getVariable("Database::database"), 
            Con::getVariable("Database::host"), 
            Con::getVariable("Database::user"), 
            Con::getVariable("Database::password")))
         {
            Con::printf("Unable to reconnect with connect()");
         }
      }

      switch(event->type2)
      {
         case GenericQuery:
            Con::printf("processGenericQuery");
            processGenericQuery((GenericQueryRequest*)event);
            break;
      }
   }
   catch(...)
   {
      Con::printf("Database thread threw an unhandled exception.");
   }
}

void DatabaseInterface::processGenericQuery(GenericQueryRequest* pRequest)
{
   // Create the reply message and initialize it from the request
   GenericQueryReply reply;
   reply.pQueryCallback = pRequest->pQueryCallback;
   reply.pRecordSet = pRequest->pRecordSet;

   // Create the query
   mysqlpp::Query query = m_con.query();
   query << pRequest->pSQLString;

   // Free up the memory for this query
   delete pRequest->pSQLString;

   // Execute the query
   mysqlpp::ResUse res = query.use();

   if (res)
   {
      try
      {
         mysqlpp::Row row;

         // Set the field count
         reply.pRecordSet->setFieldCount(res.columns());

         // Loop through all of the columns and set them in the recordset
         for(int x = 0; x < res.columns(); x++)
         {
            reply.pRecordSet->setFieldName(x, res.field_name(x).c_str());
         }

         // Loop through all of the rows and set the record field values
         while (row = res.fetch_row()) 
         {
            reply.pRecordSet->addRecord();
            for(int x = 0; x < res.columns(); x++)
            {
               reply.pRecordSet->setFieldValue(x, row.at(x).c_str());
            }
         }
      }
      catch(...)
      {
         // Hard error
         Con::errorf("DatabaseInterface::processEvent exception!");
      }
   }
   else
   {
      // Soft error
      Con::errorf("DatabaseInterface::processEvent error: %s", query.error().c_str());
   }

   Game->postEvent(reply);
}

//-~-~-~-~-~-~-~-~-~-~-~-~-~-~
   // Additional implementation
protected:

...

   /// Process a generic query
   void processGenericQuery(GenericQueryRequest* pRequest);

function testDatabase()
{
   // Save the query Id for later use if you need
   %queryId = query("Select * from Toon", testDatabaseCallback);
}

function testDatabaseCallback(%recordSet)
{
   // Compare the queryId with the one 
   echo(%recordSet.getQueryId());

   while(%recordSet.nextRecord())
   {
      // Dump all of the fields
      for(%x = 0; %x < %recordSet.getFieldCount(); %x++)
      {
         echo(%recordSet.getFieldName(%x) SPC "=" SPC %recordSet.getFieldValue(%x));
      }

      echo(%recordSet.getFieldValue("FirstName") SPC %recordSet.getFieldValue("LastName"));
   }
   
   // Make sure to clean this up.
   %recordSet.delete();
}

void GameInterface::processDatabaseReplyEventType(DatabaseReply* event)
{
   switch(event->type2)
   {
   case GenericQuery:
      {
         GenericQueryReply* const pReply = (GenericQueryReply*)event;

         Con::executef(3, pReply->pQueryCallback, pReply->pRecordSet->getIdString());

         // Delete the callback string, but don't delete the recordset
         // because it'll be handled by the script engine.
         delete pReply->pQueryCallback;
      }
      break;
   default:
      Con::printf("Got an illegal database reply type: %i", event->type2);
      break;
   }
}

#ifndef _PLATFORMASSERT_H_
# include "platform/platformAssert.h"
#endif

#define TORQUE_DISABLE_MEMORY_MANAGER

#include <new>