/cygdrive/d/src/svn/vrut/trunk/core/src/msgqueue.h

Go to the documentation of this file.

// Name:        wx/thread.h
// Purpose:     Message queues for inter-thread communication
// Author:      Evgeniy Tarassov (Modified by Vaclav Kyba)
// Created:     2007-10-31
// RCS-ID:      $Id: msgqueue.h 670 2009-01-14 14:03:57Z kybav1 $
// Copyright:   (C) 2007 TT-Solutions SARL
// Licence:     wxWindows licence

#ifndef _WX_MSGQUEUE_VRUT_H_
#define _WX_MSGQUEUE_VRUT_H_

// ----------------------------------------------------------------------------
// headers
// ----------------------------------------------------------------------------

#include <wx/thread.h>

#if wxUSE_THREADS

#include <wx/stopwatch.h>

#include <wx/beforestd.h>
#include <deque>
#include <wx/afterstd.h>

namespace VRUT
{
       enum MessageQueueError
       {
              wxMSGQUEUE_NO_ERROR = 0, // operation completed successfully
              wxMSGQUEUE_TIMEOUT,      // no messages received before timeout expired
              wxMSGQUEUE_MISC_ERROR    // some unexpected (and fatal) error has occurred
       };

       // ---------------------------------------------------------------------------
       // Message queue allows passing message between threads.
       //
       // This class is typically used for communicating between the main and worker
       // threads. The main thread calls Post() and the worker thread calls Receive().
       //
       // For this class a message is an object of arbitrary type T. Notice that
       // typically there must be some special message indicating that the thread
       // should terminate as there is no other way to gracefully shutdown a thread
       // waiting on the message queue.
       // ---------------------------------------------------------------------------
       template <typename T>
       class MessageQueue
       {
       public:
              // The type of the messages transported by this queue
              typedef T Message;

              // Default ctor creates an initially empty queue
              MessageQueue()
                 : m_conditionNotEmpty(m_mutex), hPrioMesCount(0)
              {
              }

              MessageQueueError Post(const Message& msg, bool highPrio = false)
              {
                     wxMutexLocker locker(m_mutex);

                     wxCHECK( locker.IsOk(), wxMSGQUEUE_MISC_ERROR );

                     if (m_messages.size() > 0xffffff)
                            return wxMSGQUEUE_MISC_ERROR;

                     if (highPrio)
                     {
                            typename std::list<T>::iterator normMes = m_messages.begin();
                            for (size_t i=0; i<hPrioMesCount; i++)
                                   normMes++;
                            m_messages.insert(normMes, msg);
                            hPrioMesCount++;
                     }
                     else
                            m_messages.push_back(msg);

                     m_conditionNotEmpty.Signal();

                     return wxMSGQUEUE_NO_ERROR;
              }

              // Wait no more than timeout milliseconds until a message becomes available.
              //
              // Setting timeout to 0 is equivalent to an infinite timeout. See Receive().
              T ReceiveTimeout(long timeout, MessageQueueError & err)
              {
                     err = wxMSGQUEUE_MISC_ERROR;
                     if (!IsOk())
                            return T();

                     wxMutexLocker locker(m_mutex);

                     if (!locker.IsOk())
                            return T();

                     const wxMilliClock_t waitUntil = wxGetLocalTimeMillis() + timeout;
                     while ( m_messages.empty() )
                     {
                            wxCondError result = m_conditionNotEmpty.WaitTimeout(timeout);

                            if ( result == wxCOND_NO_ERROR )
                                   continue;

                            if (result != wxCOND_TIMEOUT)
                                   return T();

                            const wxMilliClock_t now = wxGetLocalTimeMillis();

                            if ( now >= waitUntil )
                            {
                                   err = wxMSGQUEUE_TIMEOUT;
                                   return T();
                            }

                            timeout = (waitUntil - now).ToLong();
                            wxASSERT(timeout > 0);
                     }

                     T ret = m_messages.front();
                     m_messages.pop_front();
                     if (hPrioMesCount>0)
                            hPrioMesCount--;

                     err = wxMSGQUEUE_NO_ERROR;
                     return ret;
              }

              T CheckFirst() const
              {
                     if (!IsOk())
                            return T();

                     wxMutexLocker locker(m_mutex);

                     if (!locker.IsOk() || m_messages.empty())
                            return T();

                     return m_messages.front();
              }

              //TODO: get last added message (not necessarily last in queue)
              T GetLast(bool remove = false)
              {
                     if (!IsOk())
                            return T();

                     wxMutexLocker locker(m_mutex);

                     if (!locker.IsOk() || m_messages.empty())
                            return T();

                     if (remove)
                     {
                            //TODO: not affecting priority count
                            T ret = m_messages.back();
                            m_messages.pop_back();
                            return ret;
                     }
                     else
                            return m_messages.back();
              }

              // Return false only if there was a fatal error in ctor
              bool IsOk() const
              {
                     return m_conditionNotEmpty.IsOk();
              }

              size_t GetSize() const
              {
                     wxMutexLocker locker(m_mutex);
                     return m_messages.size();
              }

              void Clear()
              {
                     wxMutexLocker locker(m_mutex);
                     m_messages.clear();
              }

       private:
              // Disable copy ctor and assignment operator
              MessageQueue(const MessageQueue<T>& rhs);
              MessageQueue<T>& operator=(const MessageQueue<T>& rhs);

              mutable wxMutex m_mutex;
              wxCondition     m_conditionNotEmpty;

              std::list<T>   m_messages;
              size_t hPrioMesCount;
       };
};
#endif // wxUSE_THREADS

#endif // _WX_MSGQUEUE_VRUT_H_

---