1/* 2 * libjingle 3 * Copyright 2004--2005, Google Inc. 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions are met: 7 * 8 * 1. Redistributions of source code must retain the above copyright notice, 9 * this list of conditions and the following disclaimer. 10 * 2. Redistributions in binary form must reproduce the above copyright notice, 11 * this list of conditions and the following disclaimer in the documentation 12 * and/or other materials provided with the distribution. 13 * 3. The name of the author may not be used to endorse or promote products 14 * derived from this software without specific prior written permission. 15 * 16 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED 17 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF 18 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO 19 * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 20 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 21 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; 22 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, 23 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR 24 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF 25 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 26 */ 27 28#ifndef TALK_BASE_MESSAGEQUEUE_H_ 29#define TALK_BASE_MESSAGEQUEUE_H_ 30 31#include <string.h> 32 33#include <algorithm> 34#include <list> 35#include <queue> 36#include <vector> 37 38#include "talk/base/basictypes.h" 39#include "talk/base/constructormagic.h" 40#include "talk/base/criticalsection.h" 41#include "talk/base/messagehandler.h" 42#include "talk/base/scoped_ptr.h" 43#include "talk/base/scoped_ref_ptr.h" 44#include "talk/base/sigslot.h" 45#include "talk/base/socketserver.h" 46#include "talk/base/timeutils.h" 47 48namespace talk_base { 49 50struct Message; 51class MessageQueue; 52 53// MessageQueueManager does cleanup of of message queues 54 55class MessageQueueManager { 56 public: 57 static void Add(MessageQueue *message_queue); 58 static void Remove(MessageQueue *message_queue); 59 static void Clear(MessageHandler *handler); 60 61 // For testing purposes, we expose whether or not the MessageQueueManager 62 // instance has been initialized. It has no other use relative to the rest of 63 // the functions of this class, which auto-initialize the underlying 64 // MessageQueueManager instance when necessary. 65 static bool IsInitialized(); 66 67 private: 68 static MessageQueueManager* Instance(); 69 70 MessageQueueManager(); 71 ~MessageQueueManager(); 72 73 void AddInternal(MessageQueue *message_queue); 74 void RemoveInternal(MessageQueue *message_queue); 75 void ClearInternal(MessageHandler *handler); 76 77 static MessageQueueManager* instance_; 78 // This list contains all live MessageQueues. 79 std::vector<MessageQueue *> message_queues_; 80 CriticalSection crit_; 81}; 82 83// Derive from this for specialized data 84// App manages lifetime, except when messages are purged 85 86class MessageData { 87 public: 88 MessageData() {} 89 virtual ~MessageData() {} 90}; 91 92template <class T> 93class TypedMessageData : public MessageData { 94 public: 95 explicit TypedMessageData(const T& data) : data_(data) { } 96 const T& data() const { return data_; } 97 T& data() { return data_; } 98 private: 99 T data_; 100}; 101 102// Like TypedMessageData, but for pointers that require a delete. 103template <class T> 104class ScopedMessageData : public MessageData { 105 public: 106 explicit ScopedMessageData(T* data) : data_(data) { } 107 const scoped_ptr<T>& data() const { return data_; } 108 scoped_ptr<T>& data() { return data_; } 109 private: 110 scoped_ptr<T> data_; 111}; 112 113// Like ScopedMessageData, but for reference counted pointers. 114template <class T> 115class ScopedRefMessageData : public MessageData { 116 public: 117 explicit ScopedRefMessageData(T* data) : data_(data) { } 118 const scoped_refptr<T>& data() const { return data_; } 119 scoped_refptr<T>& data() { return data_; } 120 private: 121 scoped_refptr<T> data_; 122}; 123 124template<class T> 125inline MessageData* WrapMessageData(const T& data) { 126 return new TypedMessageData<T>(data); 127} 128 129template<class T> 130inline const T& UseMessageData(MessageData* data) { 131 return static_cast< TypedMessageData<T>* >(data)->data(); 132} 133 134template<class T> 135class DisposeData : public MessageData { 136 public: 137 explicit DisposeData(T* data) : data_(data) { } 138 virtual ~DisposeData() { delete data_; } 139 private: 140 T* data_; 141}; 142 143const uint32 MQID_ANY = static_cast<uint32>(-1); 144const uint32 MQID_DISPOSE = static_cast<uint32>(-2); 145 146// No destructor 147 148struct Message { 149 Message() { 150 memset(this, 0, sizeof(*this)); 151 } 152 inline bool Match(MessageHandler* handler, uint32 id) const { 153 return (handler == NULL || handler == phandler) 154 && (id == MQID_ANY || id == message_id); 155 } 156 MessageHandler *phandler; 157 uint32 message_id; 158 MessageData *pdata; 159 uint32 ts_sensitive; 160}; 161 162typedef std::list<Message> MessageList; 163 164// DelayedMessage goes into a priority queue, sorted by trigger time. Messages 165// with the same trigger time are processed in num_ (FIFO) order. 166 167class DelayedMessage { 168 public: 169 DelayedMessage(int delay, uint32 trigger, uint32 num, const Message& msg) 170 : cmsDelay_(delay), msTrigger_(trigger), num_(num), msg_(msg) { } 171 172 bool operator< (const DelayedMessage& dmsg) const { 173 return (dmsg.msTrigger_ < msTrigger_) 174 || ((dmsg.msTrigger_ == msTrigger_) && (dmsg.num_ < num_)); 175 } 176 177 int cmsDelay_; // for debugging 178 uint32 msTrigger_; 179 uint32 num_; 180 Message msg_; 181}; 182 183class MessageQueue { 184 public: 185 explicit MessageQueue(SocketServer* ss = NULL); 186 virtual ~MessageQueue(); 187 188 SocketServer* socketserver() { return ss_; } 189 void set_socketserver(SocketServer* ss); 190 191 // Note: The behavior of MessageQueue has changed. When a MQ is stopped, 192 // futher Posts and Sends will fail. However, any pending Sends and *ready* 193 // Posts (as opposed to unexpired delayed Posts) will be delivered before 194 // Get (or Peek) returns false. By guaranteeing delivery of those messages, 195 // we eliminate the race condition when an MessageHandler and MessageQueue 196 // may be destroyed independently of each other. 197 virtual void Quit(); 198 virtual bool IsQuitting(); 199 virtual void Restart(); 200 201 // Get() will process I/O until: 202 // 1) A message is available (returns true) 203 // 2) cmsWait seconds have elapsed (returns false) 204 // 3) Stop() is called (returns false) 205 virtual bool Get(Message *pmsg, int cmsWait = kForever, 206 bool process_io = true); 207 virtual bool Peek(Message *pmsg, int cmsWait = 0); 208 virtual void Post(MessageHandler *phandler, uint32 id = 0, 209 MessageData *pdata = NULL, bool time_sensitive = false); 210 virtual void PostDelayed(int cmsDelay, MessageHandler *phandler, 211 uint32 id = 0, MessageData *pdata = NULL) { 212 return DoDelayPost(cmsDelay, TimeAfter(cmsDelay), phandler, id, pdata); 213 } 214 virtual void PostAt(uint32 tstamp, MessageHandler *phandler, 215 uint32 id = 0, MessageData *pdata = NULL) { 216 return DoDelayPost(TimeUntil(tstamp), tstamp, phandler, id, pdata); 217 } 218 virtual void Clear(MessageHandler *phandler, uint32 id = MQID_ANY, 219 MessageList* removed = NULL); 220 virtual void Dispatch(Message *pmsg); 221 virtual void ReceiveSends(); 222 223 // Amount of time until the next message can be retrieved 224 virtual int GetDelay(); 225 226 bool empty() const { return size() == 0u; } 227 size_t size() const { 228 CritScope cs(&crit_); // msgq_.size() is not thread safe. 229 return msgq_.size() + dmsgq_.size() + (fPeekKeep_ ? 1u : 0u); 230 } 231 232 // Internally posts a message which causes the doomed object to be deleted 233 template<class T> void Dispose(T* doomed) { 234 if (doomed) { 235 Post(NULL, MQID_DISPOSE, new DisposeData<T>(doomed)); 236 } 237 } 238 239 // When this signal is sent out, any references to this queue should 240 // no longer be used. 241 sigslot::signal0<> SignalQueueDestroyed; 242 243 protected: 244 class PriorityQueue : public std::priority_queue<DelayedMessage> { 245 public: 246 container_type& container() { return c; } 247 void reheap() { make_heap(c.begin(), c.end(), comp); } 248 }; 249 250 void DoDelayPost(int cmsDelay, uint32 tstamp, MessageHandler *phandler, 251 uint32 id, MessageData* pdata); 252 253 // The SocketServer is not owned by MessageQueue. 254 SocketServer* ss_; 255 // If a server isn't supplied in the constructor, use this one. 256 scoped_ptr<SocketServer> default_ss_; 257 bool fStop_; 258 bool fPeekKeep_; 259 Message msgPeek_; 260 MessageList msgq_; 261 PriorityQueue dmsgq_; 262 uint32 dmsgq_next_num_; 263 mutable CriticalSection crit_; 264 265 private: 266 DISALLOW_COPY_AND_ASSIGN(MessageQueue); 267}; 268 269} // namespace talk_base 270 271#endif // TALK_BASE_MESSAGEQUEUE_H_ 272