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