include/wx/msgqueue.h

   1 ///////////////////////////////////////////////////////////////////////////////
   2 // Name:        wx/thread.h
   3 // Purpose:     Message queues for inter-thread communication
   4 // Author:      Evgeniy Tarassov
   5 // Created:     2007-10-31
   6 // RCS-ID:      $Id$
   7 // Copyright:   (C) 2007 TT-Solutions SARL
   8 // Licence:     wxWindows licence
   9 ///////////////////////////////////////////////////////////////////////////////
  10
  11 #ifndef _WX_MSGQUEUE_H_
  12 #define _WX_MSGQUEUE_H_
  13
  14 // ----------------------------------------------------------------------------
  15 // headers
  16 // ----------------------------------------------------------------------------
  17
  18 #include "wx/thread.h"
  19
  20 #if wxUSE_THREADS
  21
  22 #include "wx/beforestd.h"
  23 #include <queue>
  24 #include "wx/afterstd.h"
  25
  26 enum wxMessageQueueError
  27 {
  28     wxMSGQUEUE_NO_ERROR = 0, // operation completed successfully
  29     wxMSGQUEUE_TIMEOUT,      // no messages received before timeout expired
  30     wxMSGQUEUE_MISC_ERROR    // some unexpected (and fatal) error has occurred
  31 };
  32
  33 // ---------------------------------------------------------------------------
  34 // Message queue allows passing message between threads.
  35 //
  36 // This class is typically used for communicating between the main and worker
  37 // threads. The main thread calls Post() and the worker thread calls Receive().
  38 //
  39 // For this class a message is an object of arbitrary type T. Notice that
  40 // typically there must be some special message indicating that the thread
  41 // should terminate as there is no other way to gracefully shutdown a thread
  42 // waiting on the message queue.
  43 // ---------------------------------------------------------------------------
  44 template <typename T>
  45 class wxMessageQueue
  46 {
  47 public:
  48     // The type of the messages transported by this queue
  49     typedef T Message;
  50
  51     // Default ctor creates an initially empty queue
  52     wxMessageQueue()
  53        : m_conditionNotEmpty(m_mutex)
  54     {
  55     }
  56
  57     // Add a message to this queue and signal the threads waiting for messages.
  58     //
  59     // This method is safe to call from multiple threads in parallel.
  60     wxMessageQueueError Post(const Message& msg)
  61     {
  62         wxMutexLocker locker(m_mutex);
  63
  64         wxCHECK( locker.IsOk(), wxMSGQUEUE_MISC_ERROR );
  65
  66         m_messages.push(msg);
  67
  68         m_conditionNotEmpty.Signal();
  69
  70         return wxMSGQUEUE_NO_ERROR;
  71     }
  72
  73     // Wait no more than timeout milliseconds until a message becomes available.
  74     //
  75     // Setting timeout to 0 is equivalent to an infinite timeout. See Receive().
  76     wxMessageQueueError ReceiveTimeout(long timeout, T& msg)
  77     {
  78         wxCHECK( IsOk(), wxMSGQUEUE_MISC_ERROR );
  79
  80         wxMutexLocker locker(m_mutex);
  81
  82         wxCHECK( locker.IsOk(), wxMSGQUEUE_MISC_ERROR );
  83
  84         const wxMilliClock_t waitUntil = wxGetLocalTimeMillis() + timeout;
  85         while ( m_messages.empty() )
  86         {
  87             wxCondError result = m_conditionNotEmpty.WaitTimeout(timeout);
  88
  89             if ( result == wxCOND_NO_ERROR )
  90                 continue;
  91
  92             wxCHECK( result == wxCOND_TIMEOUT, wxMSGQUEUE_MISC_ERROR );
  93
  94             const wxMilliClock_t now = wxGetLocalTimeMillis();
  95
  96             if ( now >= waitUntil )
  97                 return wxMSGQUEUE_TIMEOUT;
  98
  99             timeout = (waitUntil - now).ToLong();
 100             wxASSERT(timeout > 0);
 101         }
 102
 103         msg = m_messages.front();
 104         m_messages.pop();
 105
 106         return wxMSGQUEUE_NO_ERROR;
 107     }
 108
 109     // Same as ReceiveTimeout() but waits for as long as it takes for a message
 110     // to become available (so it can't return wxMSGQUEUE_TIMEOUT)
 111     wxMessageQueueError Receive(T& msg)
 112     {
 113         wxCHECK( IsOk(), wxMSGQUEUE_MISC_ERROR );
 114
 115         wxMutexLocker locker(m_mutex);
 116
 117         wxCHECK( locker.IsOk(), wxMSGQUEUE_MISC_ERROR );
 118
 119         while ( m_messages.empty() )
 120         {
 121             wxCondError result = m_conditionNotEmpty.Wait();
 122
 123             wxCHECK( result == wxCOND_NO_ERROR, wxMSGQUEUE_MISC_ERROR );
 124         }
 125
 126         msg = m_messages.front();
 127         m_messages.pop();
 128
 129         return wxMSGQUEUE_NO_ERROR;
 130     }
 131
 132     // Return false only if there was a fatal error in ctor
 133     bool IsOk() const
 134     {
 135         return m_conditionNotEmpty.IsOk();
 136     }
 137
 138 private:
 139     // Disable copy ctor and assignment operator
 140     wxMessageQueue(const wxMessageQueue<T>& rhs);
 141     wxMessageQueue<T>& operator=(const wxMessageQueue<T>& rhs);
 142
 143     mutable wxMutex m_mutex;
 144     wxCondition     m_conditionNotEmpty;
 145
 146     std::queue<T>   m_messages;
 147 };
 148
 149 #endif // wxUSE_THREADS
 150
 151 #endif // _WX_MSGQUEUE_H_