Home | History | Annotate | Download | only in base 
      1 /*
      2  *  Copyright 2004 The WebRTC Project Authors. All rights reserved.
      3  *
      4  *  Use of this source code is governed by a BSD-style license
      5  *  that can be found in the LICENSE file in the root of the source
      6  *  tree. An additional intellectual property rights grant can be found
      7  *  in the file PATENTS.  All contributing project authors may
      8  *  be found in the AUTHORS file in the root of the source tree.
      9  */
     10 
     11 #ifndef WEBRTC_BASE_MESSAGEQUEUE_H_
     12 #define WEBRTC_BASE_MESSAGEQUEUE_H_
     13 
     14 #include <string.h>
     15 
     16 #include <algorithm>
     17 #include <list>
     18 #include <queue>
     19 #include <vector>
     20 
     21 #include "webrtc/base/basictypes.h"
     22 #include "webrtc/base/constructormagic.h"
     23 #include "webrtc/base/criticalsection.h"
     24 #include "webrtc/base/messagehandler.h"
     25 #include "webrtc/base/scoped_ptr.h"
     26 #include "webrtc/base/scoped_ref_ptr.h"
     27 #include "webrtc/base/sigslot.h"
     28 #include "webrtc/base/socketserver.h"
     29 #include "webrtc/base/timeutils.h"
     30 
     31 namespace rtc {
     32 
     33 struct Message;
     34 class MessageQueue;
     35 
     36 // MessageQueueManager does cleanup of of message queues
     37 
     38 class MessageQueueManager {
     39  public:
     40   static void Add(MessageQueue *message_queue);
     41   static void Remove(MessageQueue *message_queue);
     42   static void Clear(MessageHandler *handler);
     43 
     44   // For testing purposes, we expose whether or not the MessageQueueManager
     45   // instance has been initialized. It has no other use relative to the rest of
     46   // the functions of this class, which auto-initialize the underlying
     47   // MessageQueueManager instance when necessary.
     48   static bool IsInitialized();
     49 
     50  private:
     51   static MessageQueueManager* Instance();
     52 
     53   MessageQueueManager();
     54   ~MessageQueueManager();
     55 
     56   void AddInternal(MessageQueue *message_queue);
     57   void RemoveInternal(MessageQueue *message_queue);
     58   void ClearInternal(MessageHandler *handler);
     59 
     60   static MessageQueueManager* instance_;
     61   // This list contains all live MessageQueues.
     62   std::vector<MessageQueue *> message_queues_;
     63   CriticalSection crit_;
     64 };
     65 
     66 // Derive from this for specialized data
     67 // App manages lifetime, except when messages are purged
     68 
     69 class MessageData {
     70  public:
     71   MessageData() {}
     72   virtual ~MessageData() {}
     73 };
     74 
     75 template <class T>
     76 class TypedMessageData : public MessageData {
     77  public:
     78   explicit TypedMessageData(const T& data) : data_(data) { }
     79   const T& data() const { return data_; }
     80   T& data() { return data_; }
     81  private:
     82   T data_;
     83 };
     84 
     85 // Like TypedMessageData, but for pointers that require a delete.
     86 template <class T>
     87 class ScopedMessageData : public MessageData {
     88  public:
     89   explicit ScopedMessageData(T* data) : data_(data) { }
     90   const scoped_ptr<T>& data() const { return data_; }
     91   scoped_ptr<T>& data() { return data_; }
     92  private:
     93   scoped_ptr<T> data_;
     94 };
     95 
     96 // Like ScopedMessageData, but for reference counted pointers.
     97 template <class T>
     98 class ScopedRefMessageData : public MessageData {
     99  public:
    100   explicit ScopedRefMessageData(T* data) : data_(data) { }
    101   const scoped_refptr<T>& data() const { return data_; }
    102   scoped_refptr<T>& data() { return data_; }
    103  private:
    104   scoped_refptr<T> data_;
    105 };
    106 
    107 template<class T>
    108 inline MessageData* WrapMessageData(const T& data) {
    109   return new TypedMessageData<T>(data);
    110 }
    111 
    112 template<class T>
    113 inline const T& UseMessageData(MessageData* data) {
    114   return static_cast< TypedMessageData<T>* >(data)->data();
    115 }
    116 
    117 template<class T>
    118 class DisposeData : public MessageData {
    119  public:
    120   explicit DisposeData(T* data) : data_(data) { }
    121   virtual ~DisposeData() { delete data_; }
    122  private:
    123   T* data_;
    124 };
    125 
    126 const uint32_t MQID_ANY = static_cast<uint32_t>(-1);
    127 const uint32_t MQID_DISPOSE = static_cast<uint32_t>(-2);
    128 
    129 // No destructor
    130 
    131 struct Message {
    132   Message() {
    133     memset(this, 0, sizeof(*this));
    134   }
    135   inline bool Match(MessageHandler* handler, uint32_t id) const {
    136     return (handler == NULL || handler == phandler)
    137            && (id == MQID_ANY || id == message_id);
    138   }
    139   MessageHandler *phandler;
    140   uint32_t message_id;
    141   MessageData *pdata;
    142   uint32_t ts_sensitive;
    143 };
    144 
    145 typedef std::list<Message> MessageList;
    146 
    147 // DelayedMessage goes into a priority queue, sorted by trigger time.  Messages
    148 // with the same trigger time are processed in num_ (FIFO) order.
    149 
    150 class DelayedMessage {
    151  public:
    152   DelayedMessage(int delay, uint32_t trigger, uint32_t num, const Message& msg)
    153       : cmsDelay_(delay), msTrigger_(trigger), num_(num), msg_(msg) {}
    154 
    155   bool operator< (const DelayedMessage& dmsg) const {
    156     return (dmsg.msTrigger_ < msTrigger_)
    157            || ((dmsg.msTrigger_ == msTrigger_) && (dmsg.num_ < num_));
    158   }
    159 
    160   int cmsDelay_;  // for debugging
    161   uint32_t msTrigger_;
    162   uint32_t num_;
    163   Message msg_;
    164 };
    165 
    166 class MessageQueue {
    167  public:
    168   static const int kForever = -1;
    169 
    170   explicit MessageQueue(SocketServer* ss = NULL);
    171   virtual ~MessageQueue();
    172 
    173   SocketServer* socketserver() { return ss_; }
    174   void set_socketserver(SocketServer* ss);
    175 
    176   // Note: The behavior of MessageQueue has changed.  When a MQ is stopped,
    177   // futher Posts and Sends will fail.  However, any pending Sends and *ready*
    178   // Posts (as opposed to unexpired delayed Posts) will be delivered before
    179   // Get (or Peek) returns false.  By guaranteeing delivery of those messages,
    180   // we eliminate the race condition when an MessageHandler and MessageQueue
    181   // may be destroyed independently of each other.
    182   virtual void Quit();
    183   virtual bool IsQuitting();
    184   virtual void Restart();
    185 
    186   // Get() will process I/O until:
    187   //  1) A message is available (returns true)
    188   //  2) cmsWait seconds have elapsed (returns false)
    189   //  3) Stop() is called (returns false)
    190   virtual bool Get(Message *pmsg, int cmsWait = kForever,
    191                    bool process_io = true);
    192   virtual bool Peek(Message *pmsg, int cmsWait = 0);
    193   virtual void Post(MessageHandler* phandler,
    194                     uint32_t id = 0,
    195                     MessageData* pdata = NULL,
    196                     bool time_sensitive = false);
    197   virtual void PostDelayed(int cmsDelay,
    198                            MessageHandler* phandler,
    199                            uint32_t id = 0,
    200                            MessageData* pdata = NULL);
    201   virtual void PostAt(uint32_t tstamp,
    202                       MessageHandler* phandler,
    203                       uint32_t id = 0,
    204                       MessageData* pdata = NULL);
    205   virtual void Clear(MessageHandler* phandler,
    206                      uint32_t id = MQID_ANY,
    207                      MessageList* removed = NULL);
    208   virtual void Dispatch(Message *pmsg);
    209   virtual void ReceiveSends();
    210 
    211   // Amount of time until the next message can be retrieved
    212   virtual int GetDelay();
    213 
    214   bool empty() const { return size() == 0u; }
    215   size_t size() const {
    216     CritScope cs(&crit_);  // msgq_.size() is not thread safe.
    217     return msgq_.size() + dmsgq_.size() + (fPeekKeep_ ? 1u : 0u);
    218   }
    219 
    220   // Internally posts a message which causes the doomed object to be deleted
    221   template<class T> void Dispose(T* doomed) {
    222     if (doomed) {
    223       Post(NULL, MQID_DISPOSE, new DisposeData<T>(doomed));
    224     }
    225   }
    226 
    227   // When this signal is sent out, any references to this queue should
    228   // no longer be used.
    229   sigslot::signal0<> SignalQueueDestroyed;
    230 
    231  protected:
    232   class PriorityQueue : public std::priority_queue<DelayedMessage> {
    233    public:
    234     container_type& container() { return c; }
    235     void reheap() { make_heap(c.begin(), c.end(), comp); }
    236   };
    237 
    238   void DoDelayPost(int cmsDelay,
    239                    uint32_t tstamp,
    240                    MessageHandler* phandler,
    241                    uint32_t id,
    242                    MessageData* pdata);
    243 
    244   // The SocketServer is not owned by MessageQueue.
    245   SocketServer* ss_;
    246   // If a server isn't supplied in the constructor, use this one.
    247   scoped_ptr<SocketServer> default_ss_;
    248   bool fStop_;
    249   bool fPeekKeep_;
    250   Message msgPeek_;
    251   MessageList msgq_;
    252   PriorityQueue dmsgq_;
    253   uint32_t dmsgq_next_num_;
    254   mutable CriticalSection crit_;
    255 
    256  private:
    257   RTC_DISALLOW_COPY_AND_ASSIGN(MessageQueue);
    258 };
    259 
    260 }  // namespace rtc
    261 
    262 #endif  // WEBRTC_BASE_MESSAGEQUEUE_H_
    263