etevent.h

Go to the documentation of this file.

/*
* Copyright (c) 2019 Sprint
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
 
#ifndef __ETEVENT_H
#define __ETEVENT_H
 
#include <unistd.h>
#include <sys/syscall.h>
 
#include "ebase.h"
#include "etbasic.h"
#include "eerror.h"
#include "egetopt.h"
#include "eshmem.h"
#include "esynch.h"
#include "esynch2.h"
#include "etimer.h"
 
 
 
DECLARE_ERROR(EThreadQueueBaseError_NotOpenForWriting);
DECLARE_ERROR(EThreadQueueBaseError_NotOpenForReading);
DECLARE_ERROR(EThreadQueueBaseError_MultipleReadersNotAllowed);
DECLARE_ERROR(EThreadQueuePublicError_UnInitialized);
 
DECLARE_ERROR_ADVANCED(EThreadTimerError_UnableToInitialize);
DECLARE_ERROR_ADVANCED(EThreadTimerError_NotInitialized);
DECLARE_ERROR_ADVANCED(EThreadTimerError_UnableToStart);
DECLARE_ERROR_ADVANCED(EThreadTimerError_UnableToRegisterTimerHandler);
 
 
 
class EThreadEventMessageDataBase
{
public:
   EThreadEventMessageDataBase() : m_msgid() {}
   EThreadEventMessageDataBase(UInt msgid) : m_msgid(msgid) {}
 
   UInt getMessageId() { return m_msgid; }
   EThreadEventMessageDataBase &setMessageId(UInt msgid) { m_msgid = msgid; return *this; }
   virtual pVoid getVoidPtr() = 0;
   virtual Void setVoidPtr(pVoid p) = 0;
 
   ETimer &getTimer() { return m_timer; }
 
private:   
   ETimer m_timer;
   UInt m_msgid;
};
 
class EThreadEventMessageData : public EThreadEventMessageDataBase
{
public:
   typedef union {
      pVoid       voidptr;
      LongLong    int64;
      Long        int32[sizeof(pVoid) / sizeof(Long)];
      Short       int16[sizeof(pVoid) / sizeof(Short)];
      Char        int8[sizeof(pVoid) / sizeof(Char)];
      ULongLong   uint64;
      ULong       uint32[sizeof(pVoid) / sizeof(ULong)];
      UShort      uint16[sizeof(pVoid) / sizeof(UShort)];
      UChar       uint8[sizeof(pVoid) / sizeof(UChar)];
   } DataUnion;
 
   EThreadEventMessageData() : EThreadEventMessageDataBase(), m_data() {}
   EThreadEventMessageData(UInt msgid) : EThreadEventMessageDataBase(msgid), m_data() {}
   EThreadEventMessageData(UInt msgid, EThreadEventMessageData &data) : EThreadEventMessageDataBase(msgid), m_data(data.data()) {}
   EThreadEventMessageData(UInt msgid, pVoid v) : EThreadEventMessageDataBase(msgid), m_data{.voidptr=v} {}
   EThreadEventMessageData(UInt msgid, LongLong v) : EThreadEventMessageDataBase(msgid), m_data{.int64=v} {}
   EThreadEventMessageData(UInt msgid, ULongLong v) : EThreadEventMessageDataBase(msgid), m_data{.uint64=v} {}
   EThreadEventMessageData(UInt msgid, Long v1, Long v2) : EThreadEventMessageDataBase(msgid), m_data{.int32={v1,v2}} {}
   EThreadEventMessageData(UInt msgid, ULong v1, ULong v2) : EThreadEventMessageDataBase(msgid), m_data{.uint32={v1,v2}} {}
   EThreadEventMessageData(UInt msgid, Short v1, Short v2, Short v3, Short v4) : EThreadEventMessageDataBase(msgid), m_data{.int16={v1,v2,v3,v4}} {}
   EThreadEventMessageData(UInt msgid, UShort v1, UShort v2, UShort v3, UShort v4) : EThreadEventMessageDataBase(msgid), m_data{.uint16={v1,v2,v3,v4}} {}
   EThreadEventMessageData(UInt msgid, Char v1, Char v2, Char v3, Char v4, Char v5, Char v6, Char v7, Char v8) : EThreadEventMessageDataBase(msgid), m_data{.int8={v1,v2,v3,v4,v5,v6,v7,v8}} {}
   EThreadEventMessageData(UInt msgid, UChar v1, UChar v2, UChar v3, UChar v4, UChar v5, UChar v6, UChar v7, UChar v8) : EThreadEventMessageDataBase(msgid), m_data{.uint8={v1,v2,v3,v4,v5,v6,v7,v8}} {}
 
   DataUnion &data() { return m_data; }
 
   pVoid getVoidPtr() { return m_data.voidptr; }
   Void setVoidPtr(pVoid p) { m_data.voidptr = p; }
 
private:
   DataUnion m_data;
};
 
class _EThreadEventMessageBase
{
public:
   virtual ~_EThreadEventMessageBase() {}
};
 
template <class T>
class EThreadEventMessageBase : public _EThreadEventMessageBase
{
public:
   EThreadEventMessageBase()
      : m_data()
   {
   }
   EThreadEventMessageBase(const T &data)
      : m_data( data )
   {
   }
   virtual ~EThreadEventMessageBase()
   {
   }
   T &operator=(const T &data)
   {
      m_data = data;
      return *this;
   }
   T &data()
   {
      return m_data;
   }
   UInt getMessageId() { return m_data.getMessageId(); }
   EThreadEventMessageBase &setMessageId(UInt msgid) { m_data.setMessageid( msgid ); return *this; }
   pVoid getVoidPtr() { return m_data.getVoidPtr(); }
   Void setVoidPtr(pVoid p) { m_data.setVoidPtr(p); }
   ETimer &getTimer() { return m_data.getTimer(); }
 
private:
   T m_data;
};
 
typedef EThreadEventMessageBase<EThreadEventMessageData> _EThreadMessage;
 
class EThreadMessage : public _EThreadMessage
{
public:
   EThreadMessage() : _EThreadMessage() {}
   EThreadMessage(UInt msgid) : _EThreadMessage(EThreadEventMessageData(msgid)) {}
   EThreadMessage(UInt msgid, EThreadEventMessageData &data) : _EThreadMessage(data) {}
   EThreadMessage(UInt msgid, pVoid v) : _EThreadMessage(EThreadEventMessageData(msgid, v)) {}
   EThreadMessage(UInt msgid, LongLong v) : _EThreadMessage(EThreadEventMessageData(msgid, v)) {}
   EThreadMessage(UInt msgid, ULongLong v) : _EThreadMessage(EThreadEventMessageData(msgid, v)) {}
   EThreadMessage(UInt msgid, Long v1, Long v2=0) : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2)) {}
   EThreadMessage(UInt msgid, ULong v1, ULong v2=0) : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2)) {}
   EThreadMessage(UInt msgid, Short v1, Short v2=0, Short v3=0, Short v4=0) : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2, v3, v4)) {}
   EThreadMessage(UInt msgid, UShort v1, UShort v2=0, UShort v3=0, UShort v4=0) : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2, v3, v4)) {}
   EThreadMessage(UInt msgid, Char v1, Char v2=0, Char v3=0, Char v4=0, Char v5=0, Char v6=0, Char v7=0, Char v8=0)
      : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2, v3, v4, v5, v6, v7, v8)) {}
   EThreadMessage(UInt msgid, UChar v1, UChar v2=0, UChar v3=0, UChar v4=0, UChar v5=0, UChar v6=0, UChar v7=0, UChar v8=0)
      : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2, v3, v4, v5, v6, v7, v8)) {}
};
 
 
enum class EThreadQueueMode
{
   ReadOnly,
   WriteOnly,
   ReadWrite
};
 
template <class T>
class EThreadQueueBase
{
public:
 
   Bool push(const T &msg, Bool wait = True)
   {
      if (m_mode == EThreadQueueMode::ReadOnly)
         throw EThreadQueueBaseError_NotOpenForWriting();
 
      if (semFree().Decrement(wait) < 0)
         return False;
 
      {
         EMutexLock l(mutex());
 
         data()[msgHead()] = msg;
         data()[msgHead()].data().getTimer().Start();
 
         msgHead()++;
 
         if (msgHead() >= msgCnt())
            msgHead() = 0;
      }
 
      semMsgs().Increment();
 
      return True;
   }
   Bool pop(T &msg, Bool wait = True)
   {
      if (m_mode == EThreadQueueMode::WriteOnly)
         throw EThreadQueueBaseError_NotOpenForReading();
 
      if (!semMsgs().Decrement(wait))
         return False;
 
      msg = data()[msgTail()++];
 
      if (msgTail() >= msgCnt())
         msgTail() = 0;
 
      semFree().Increment();
 
      return True;
   }
   Bool peek(T &msg, Bool wait = True)
   {
      if (m_mode == EThreadQueueMode::WriteOnly)
         throw EThreadQueueBaseError_NotOpenForReading();
 
      if (!semMsgs().Decrement(wait))
         return False;
 
      msg = data()[msgTail()];
 
      // since we are not pulling the message off, we need to increment
      // the semaphore to put it back where it was
      semMsgs().Increment();
 
      return True;
   }
 
   Bool isInitialized() { return m_initialized; }
   EThreadQueueMode mode() { return m_mode; }
 
protected:
 
   virtual Bool isPublic() = 0;
   virtual Int &msgCnt() = 0;
   virtual Int &msgHead() = 0;
   virtual Int &msgTail() = 0;
   virtual Bool &multipleWriters() = 0;
   virtual Int &numReaders() = 0;
   virtual Int &numWriters() = 0;
   virtual Int &refCnt() = 0;
   virtual T *data() = 0;
   virtual Void allocDataSpace(cpStr sFile, Char cId, Int nSize) = 0;
   virtual Void initMutex() = 0;
   virtual Void initSemFree(UInt initialCount) = 0;
   virtual Void initSemMsgs(UInt initialCount) = 0;
 
   virtual EMutexData &mutex() = 0;
   virtual ESemaphoreData &semMsgs() = 0;
   virtual ESemaphoreData &semFree() = 0;
 
   virtual int *getBumpPipe() = 0;
 
   EThreadQueueBase()
   {
      m_initialized = False;
   }
   ~EThreadQueueBase()
   {
   }
 
   Void init(Int nMsgCnt, Int threadId, Bool bMultipleWriters, EThreadQueueMode eMode)
   {
      m_mode = eMode;
 
      // construct the shared memory name
      Char szName[EPC_FILENAME_MAX];
      epc_sprintf_s(szName, sizeof(szName), "%d", threadId);
 
      // calcuate the space required
      int nSize = sizeof(T) * nMsgCnt;
 
      // initialize the shared memory
      allocDataSpace(szName, 'A', nSize);
 
      // initialize the control block values
      if (refCnt() == 0)
      {
         refCnt() = 0;
         numReaders() = 0;
         numWriters() = 0;
         multipleWriters() = bMultipleWriters;
         msgCnt() = nMsgCnt;
         msgHead() = 0;
         msgTail() = 0;
 
         // initialize the control mutex and semaphores
         initMutex();
         initSemFree(msgCnt());
         initSemMsgs(0);
      }
      else
      {
      }
 
      EMutexLock l(mutex());
 
      if ((eMode == EThreadQueueMode::ReadOnly || eMode == EThreadQueueMode::ReadWrite) && numReaders() > 0)
      {
         throw EThreadQueueBaseError_MultipleReadersNotAllowed();
      }
 
      refCnt()++;
      numReaders() += (eMode == EThreadQueueMode::ReadOnly || eMode == EThreadQueueMode::ReadWrite) ? 1 : 0;
      numWriters() += (eMode == EThreadQueueMode::WriteOnly || eMode == EThreadQueueMode::ReadWrite) ? 1 : 0;
 
      m_initialized = True;
   }
 
   Void destroy()
   {
      Bool destroyMutex = False;
 
      if (m_initialized)
      {
         EMutexLock l(mutex());
 
         if (refCnt() == 1)
         {
            semFree().destroy();
            semMsgs().destroy();
 
            destroyMutex = True;
         }
         else
         {
            refCnt()--;
            numReaders() -= (m_mode == EThreadQueueMode::ReadOnly || m_mode == EThreadQueueMode::ReadWrite) ? 1 : 0;
            numWriters() -= (m_mode == EThreadQueueMode::WriteOnly || m_mode == EThreadQueueMode::ReadWrite) ? 1 : 0;
         }
      }
 
      if (destroyMutex)
         mutex().destroy();
   }
 
private:
   Bool m_initialized;
   EThreadQueueMode m_mode;
};
 
 
template <class T>
class EThreadQueuePublic : public EThreadQueueBase<T>
{
   template <class TQueue, class TMessage> friend class EThreadEvent;
public:
   EThreadQueuePublic()
   {
      m_pCtrl = NULL;
      m_pData = NULL;
   }
   ~EThreadQueuePublic()
   {
   }
   Void init(Int nMsgCnt, Int threadId, Bool bMultipleWriters,
             EThreadQueueMode eMode)
   {
      EThreadQueueBase<T>::init(nMsgCnt, threadId, bMultipleWriters, eMode);
   }
 
protected:
   Bool isPublic() { return True; }
   Int &msgCnt() { return m_pCtrl->m_msgCnt; }
   Int &msgHead() { return m_pCtrl->m_head; }
   Int &msgTail() { return m_pCtrl->m_tail; }
   Bool &multipleWriters() { return m_pCtrl->m_multipleWriters; }
   Int &numReaders() { return m_pCtrl->m_numReaders; }
   Int &numWriters() { return m_pCtrl->m_numWriters; }
   Int &refCnt() { return m_pCtrl->m_refCnt; }
   T *data() { return m_pData; }
   Void allocDataSpace(cpStr sFile, Char cId, Int nSize)
   {
      m_sharedmem.init(sFile, cId, nSize + sizeof(ethreadmessagequeue_ctrl_t));
      m_pCtrl = (ethreadmessagequeue_ctrl_t *)m_sharedmem.getDataPtr();
      m_pData = (T *)(((pChar)m_sharedmem.getDataPtr()) + sizeof(ethreadmessagequeue_ctrl_t));
   }
   Void initMutex()
   {
      EMutexPublic m;
      m_pCtrl->m_mutexid = m.mutexId();
      m.detach();
   }
   Void initSemFree(UInt initialCount)
   {
      ESemaphorePublic s(initialCount);
      m_pCtrl->m_freeSemId = s.semIndex();
      s.detach();
   }
   Void initSemMsgs(UInt initialCount)
   {
      ESemaphorePublic s(initialCount);
      m_pCtrl->m_msgsSemId = s.semIndex();
      s.detach();
   }
 
   EMutexData &mutex() { return ESynchObjects::getMutex(m_pCtrl->m_mutexid); }
   ESemaphoreData &semFree() { return ESynchObjects::getSemaphore(m_pCtrl->m_freeSemId); }
   ESemaphoreData &semMsgs() { return ESynchObjects::getSemaphore(m_pCtrl->m_msgsSemId); }
 
   int *getBumpPipe() { return m_pCtrl->m_bumppipe; }
 
private:
   typedef struct
   {
      Int m_refCnt;
      Int m_numReaders;
      Int m_numWriters;
      Bool m_multipleWriters;
 
      Int m_msgCnt;
      Int m_head; // next location to write
      Int m_tail; // next location to read
 
      Int m_mutexid;
      Int m_freeSemId;
      Int m_msgsSemId;
 
      int m_bumppipe[2];
   } ethreadmessagequeue_ctrl_t;
 
   ESharedMemory m_sharedmem;
   ethreadmessagequeue_ctrl_t *m_pCtrl;
   T *m_pData;
};
 
 
template <class T>
class EThreadQueuePrivate : public EThreadQueueBase<T>
{
   template <class TQueue, class TMessage> friend class EThreadEvent;
public:
   EThreadQueuePrivate()
      : m_mutex(False), m_semFree(0, False), m_semMsgs(0, False)
   {
      m_refCnt = 0;
      m_numReaders = 0;
      m_numWriters = 0;
      m_multipleWriters = False;
 
      m_msgCnt = 0;
      m_head = 0; // next location to write
      m_tail = 0; // next location to read
      m_pData = NULL;
   }
   ~EThreadQueuePrivate()
   {
      if (m_pData)
      {
         delete[](pChar) m_pData;
         m_pData = NULL;
      }
   }
   Void init(Int nMsgCnt, Int threadId, Bool bMultipleWriters,
             EThreadQueueMode eMode)
   {
      EThreadQueueBase<T>::init(nMsgCnt, threadId, bMultipleWriters, eMode);
   }
 
protected:
   Bool isPublic() { return False; }
   Int &msgCnt() { return m_msgCnt; }
   Int &msgHead() { return m_head; }
   Int &msgTail() { return m_tail; }
   Bool &multipleWriters() { return m_multipleWriters; }
   Int &numReaders() { return m_numReaders; }
   Int &numWriters() { return m_numWriters; }
   Int &refCnt() { return m_refCnt; }
   T *data() { return m_pData; }
   Void allocDataSpace(cpStr sFile, Char cId, Int nSize)
   {
      m_pData = (EThreadMessage *)new Char[nSize];
      memset((pChar)m_pData, 0, nSize);
   }
   Void initMutex()
   {
      m_mutex.init();
   }
   Void initSemFree(UInt initialCount)
   {
      m_semFree.init(initialCount);
   }
   Void initSemMsgs(UInt initialCount)
   {
      m_semMsgs.init(initialCount);
   }
 
   EMutexData &mutex() { return m_mutex; }
   ESemaphoreData &semFree() { return m_semFree; }
   ESemaphoreData &semMsgs() { return m_semMsgs; }
 
   int *getBumpPipe() { return m_bumppipe; }
 
private:
   Int m_refCnt;
   Int m_numReaders;
   Int m_numWriters;
   Bool m_multipleWriters;
 
   Int m_msgCnt;
   Int m_head; // next location to write
   Int m_tail; // next location to read
 
   EMutexPrivate m_mutex;
   ESemaphorePrivate m_semFree;
   ESemaphorePrivate m_semMsgs;
 
   int m_bumppipe[2];
 
   T *m_pData;
};
 
 
#define EM_INIT 1
#define EM_QUIT 2
#define EM_SUSPEND 3
#define EM_TIMER 4
#define EM_SOCKETSELECT_READ 5
#define EM_SOCKETSELECT_WRITE 6
#define EM_SOCKETSELECT_ERROR 7
#define EM_SOCKETSELECT_EXCEPTION 8
#define EM_USER 10000
 
 
#define DECLARE_MESSAGE_MAP()                                  \
protected:                                                     \
   static const msgmap_t *GetThisMessageMap();                 \
   virtual const msgmap_t *GetMessageMap() const;
 
#define BEGIN_MESSAGE_MAP(theClass, baseClass)                 \
   const theClass::msgmap_t *theClass::GetMessageMap() const   \
   {                                                           \
      return GetThisMessageMap();                              \
   }                                                           \
   const theClass::msgmap_t *theClass::GetThisMessageMap()     \
   {                                                           \
      typedef baseClass TheBaseClass;                          \
      _Pragma("GCC diagnostic push")                           \
      _Pragma("GCC diagnostic ignored \"-Wpmf-conversions\"")  \
      static const msgentry_t _msgEntries[] =                  \
      {
 
#define ON_MESSAGE(id, memberFxn)                              \
         {id, (msgfxn_t)&memberFxn},
 
#define END_MESSAGE_MAP()                                      \
         {0, (msgfxn_t)NULL}                                   \
      };                                                       \
      _Pragma("GCC diagnostic pop")                            \
      static const msgmap_t msgMap =                           \
         {&TheBaseClass::GetThisMessageMap, &_msgEntries[0]};  \
      return &msgMap;                                          \
   }
 
 
class _EThreadEventBase
{
public:
   virtual Bool _sendMessage(const _EThreadEventMessageBase &msg, Bool wait = True)
   {
      return False;
   }
   virtual Void _destroyMessage(_EThreadEventMessageBase *msg)
   {
      delete msg;
   }
};
 
class EThreadEventTimer : public EStatic
{
   friend class EThreadEventTimerHandler;
   template <class TQueue, class TMessage> friend class EThreadEvent;
 
protected:
   Void init(_EThreadEventBase *thread, _EThreadEventMessageBase *msg)
   {
      m_thread = thread;
      m_msg = msg;
 
      struct sigevent sev = {};
      sev.sigev_notify = SIGEV_SIGNAL;
      sev.sigev_signo = SIGRTMIN;
      sev.sigev_value.sival_ptr = this;
      if (timer_create(CLOCK_REALTIME, &sev, &m_timer) == -1)
         throw EThreadTimerError_UnableToInitialize();
   }
 
public:
   EThreadEventTimer()
   {
      // assign the id
      m_id = atomic_inc(m_nextid);
      m_thread = NULL;
      m_msg = NULL;
      m_interval = 0;
      m_oneshot = True;
      m_timer = NULL;
   }
   EThreadEventTimer(Long milliseconds, Bool oneshot = False)
   {
      // assign the id
      m_id = atomic_inc(m_nextid);
      m_thread = NULL;
      m_msg = NULL;
      m_interval = milliseconds;
      m_oneshot = oneshot;
      m_timer = NULL;
   }      
   ~EThreadEventTimer()
   {
      destroy();
   }
   Void destroy()
   {
      if (m_timer != NULL)
      {
         stop();
         timer_delete(m_timer);
         m_timer = NULL;
      }
      if (m_msg)
      {
         if (m_thread)
            m_thread->_destroyMessage(m_msg);
         else
            delete m_msg;
         m_msg = NULL;
      }
      m_thread = NULL;
   }
   Void start()
   {
      if (m_timer == NULL)
         throw EThreadTimerError_NotInitialized();
 
      struct itimerspec its;
      its.it_value.tv_sec = m_interval / 1000;              // seconds
      its.it_value.tv_nsec = (m_interval % 1000) * 1000000; // nano-seconds
      its.it_interval.tv_sec = m_oneshot ? 0 : its.it_value.tv_sec;
      its.it_interval.tv_nsec = m_oneshot ? 0 : its.it_value.tv_nsec;
      if (timer_settime(m_timer, 0, &its, NULL) == -1)
         throw EThreadTimerError_UnableToStart();
   }
   Void stop()
   {
      if (m_timer != NULL)
      {
         struct itimerspec its;
         its.it_value.tv_sec = 0;  // seconds
         its.it_value.tv_nsec = 0; // nano-seconds
         its.it_interval.tv_sec = its.it_value.tv_sec;
         its.it_interval.tv_nsec = its.it_value.tv_nsec;
         timer_settime(m_timer, 0, &its, NULL);
      }
   }
   Long getInterval() { return m_interval; }
   Void setInterval(Long interval) { m_interval = interval; }
   Void setOneShot(Bool oneshot) { m_oneshot = oneshot; }
   Long getId() { return m_id; }
   Bool isInitialized() { return m_timer != NULL; }
 
protected:
   static void _timerHandler(int signo, siginfo_t *pinfo, void *pcontext)
   {
      EThreadEventTimer *timer = (EThreadEventTimer*)pinfo->si_value.sival_ptr;
      if (timer)
      {
         timer->m_thread->_sendMessage(*timer->m_msg);
      }
   }
 
private:
   static Long m_nextid;
 
   Long m_id;
   _EThreadEventBase *m_thread;
   _EThreadEventMessageBase *m_msg;
   Bool m_oneshot;
   Long m_interval;
   timer_t m_timer;
};
 
class EThreadEventTimerHandler : EStatic
{
public:
   EThreadEventTimerHandler() {}
   ~EThreadEventTimerHandler() {}
 
   virtual Int getInitType() { return STATIC_INIT_TYPE_THREADS; }
   Void init(EGetOpt &options)
   {
      struct sigaction sa;
      sa.sa_flags = SA_SIGINFO;
      sa.sa_sigaction = EThreadEventTimer::_timerHandler;
      sigemptyset(&sa.sa_mask);
      int signo = SIGRTMIN;
      if (sigaction(signo, &sa, NULL) == -1)
         throw EThreadTimerError_UnableToRegisterTimerHandler();
   }
   Void uninit()
   {
   }
};
 
template <class TQueue, class TMessage>
class EThreadEvent : public EThreadBasic, public _EThreadEventBase
{
public:
   typedef Void (EThreadEvent::*msgfxn_t)(TMessage &);
   typedef struct
   {
      UInt     nMessage; // message
      msgfxn_t pFn; // routine to call (or special value)
   } msgentry_t;
 
   struct msgmap_t
   {
      const msgmap_t *(*pfnGetBaseMap)();
      const msgentry_t *lpEntries;
   };
 
   EThreadEvent()
      : EThreadBasic(),
        m_tid(-1),
        m_arg(NULL),
        m_stacksize(0),
        m_suspendCnt(0),
        m_suspendSem(0)
   {
   }
   ~EThreadEvent()
   {
   }
 
   Bool sendMessage(UInt message, Bool wait = True)
   {
      TMessage msg(message);
      Bool result = m_queue.push(msg, wait);
      if (result)
         messageQueued();
      return result;
   }
   Bool sendMessage(UInt message, pVoid voidptr, Bool wait = True)
   {
      TMessage msg(message);
      msg.setVoidPtr(voidptr);
      Bool result = m_queue.push(msg, wait);
      if (result)
         messageQueued();
      return result;
   }
   Bool sendMessage(const TMessage &msg, Bool wait = True)
   {
      Bool result = m_queue.push(msg, wait);
      if (result)
         messageQueued();
      return result;
   }
 
   virtual Void init(Short appId, UShort threadId, pVoid arg, Int queueSize = 16384, Bool suspended = False, Dword stackSize = 0)
   {
      m_appId = appId;
      m_threadId = threadId;
      m_queueSize = queueSize;
      m_stacksize = stackSize;
 
      long id = m_appId * 10000 + m_threadId;
 
      m_queue.init(m_queueSize, id, True, EThreadQueueMode::ReadWrite);
 
      if (!suspended)
         start();
   }
 
   Void quit()
   {
      sendMessage(EM_QUIT);
   }
   Void start()
   {
      if (!isInitialized())
      {
         EThreadBasic::init(m_arg, m_stacksize);
         sendMessage(EM_INIT);
      }
   }
   Void suspend()
   {
      if (atomic_inc(m_suspendCnt) == 1)
         sendMessage(EM_SUSPEND);
   }
   Void resume()
   {
      if (atomic_dec(m_suspendCnt) == 0)
         m_suspendSem.Increment();
   }
   virtual Void onInit()
   {
   }
   virtual Void onQuit()
   {
   }
   virtual Void onSuspend()
   {
   }
   virtual Void onTimer(EThreadEventTimer *ptimer)
   {
   }
   Void initTimer(EThreadEventTimer &t)
   {
      TMessage *msg = new TMessage(EM_TIMER);
      msg->setVoidPtr(&t);
      t.init(this, msg);
   }
   ESemaphoreData &getMsgSemaphore()
   {
      return m_queue.semMsgs();
   }
 
protected:
   virtual const msgmap_t *GetMessageMap() const
   {
      return GetThisMessageMap();
   }
   static const msgmap_t *GetThisMessageMap()
   {
      return NULL;
   }
   int *getBumpPipe()
   {
      return m_queue.getBumpPipe();
   }
 
   virtual Void messageQueued()
   {
   }
   Bool pumpMessage(TMessage &msg, Bool wait = true)
   {
      Bool bMsg = m_queue.pop(msg, wait);
      if (bMsg)
         dispatch(msg);
 
      return bMsg;
   }
   virtual Void pumpMessages()
   {
      TMessage msg;
 
      try
      {
         while (True)
         {
            if (pumpMessage(msg))
            {
               if (msg.getMessageId() == EM_QUIT)
                  break;
               if (msg.getMessageId() == EM_SUSPEND)
                  m_suspendSem.Decrement();
            }
         }
      }
      catch (EError &e)
      {
         //std::cout << "t1 - " << e.Name() << std::endl;
         throw;
      }
      catch (...)
      {
         throw;
      }
   }
   virtual Void defaultMessageHandler(TMessage &msg)
   {
   }
   pid_t getThreadId()
   {
      if (m_tid == -1)
         m_tid = syscall(SYS_gettid);
      return m_tid;
   }
 
private:
   Dword threadProc(pVoid arg)
   {
      pumpMessages();
      return 0;
   }
 
   Bool dispatch(TMessage &msg)
   {
      Bool keepgoing = True;
      const msgmap_t *pMap;
      const msgentry_t *pEntries;
 
      if (msg.getMessageId() >= EM_USER)
      {
         // interate through each map
         for (pMap = GetMessageMap(); keepgoing && pMap && pMap->pfnGetBaseMap != NULL; pMap = (*pMap->pfnGetBaseMap)())
         {
            // interate through each entry for the map
            for (pEntries = pMap->lpEntries; pEntries->nMessage; pEntries++)
            {
               if (pEntries->nMessage == msg.getMessageId())
               {
                  (this->*pEntries->pFn)(msg);
                  keepgoing = False;
                  break;
               }
            }
         }
 
         if (pMap == NULL)
            defaultMessageHandler(msg);
      }
      else
      {
         switch (msg.getMessageId())
         {
            case EM_INIT:
            {
               onInit();
               break;
            }
            case EM_QUIT:
            {
               onQuit();
               break;
            }
            case EM_SUSPEND:
            {
               onSuspend();
               break;
            }
            case EM_TIMER:
            {
               onTimer( (EThreadEventTimer*)msg.getVoidPtr() );
               break;
            }
            default:
            {
               break;
            }
         }
      }
 
      return keepgoing;
   }
 
   Bool _sendMessage(const _EThreadEventMessageBase &msg, Bool wait)
   {
      return sendMessage( (const TMessage &)msg, wait );
   }
 
   Void _destroyMessage(_EThreadEventMessageBase *msg)
   {
      delete (TMessage*)msg;
   }
 
   pid_t m_tid;
   pVoid m_arg;
   size_t m_stacksize;
   Int m_suspendCnt;
   ESemaphorePrivate m_suspendSem;
 
   Short m_appId;
   UShort m_threadId;
   Int m_queueSize;
   TQueue m_queue;
};
 
typedef EThreadEvent<EThreadQueuePublic<EThreadMessage>,EThreadMessage> EThreadPublic;
typedef EThreadEvent<EThreadQueuePrivate<EThreadMessage>,EThreadMessage> EThreadPrivate;
 
 
#endif // #ifndef __ETEVENT_H