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
31namespace rtc {
32
33struct Message;
34class MessageQueue;
35
36// MessageQueueManager does cleanup of of message queues
37
38class 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
69class MessageData {
70 public:
71  MessageData() {}
72  virtual ~MessageData() {}
73};
74
75template <class T>
76class 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.
86template <class T>
87class 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.
97template <class T>
98class 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
107template<class T>
108inline MessageData* WrapMessageData(const T& data) {
109  return new TypedMessageData<T>(data);
110}
111
112template<class T>
113inline const T& UseMessageData(MessageData* data) {
114  return static_cast< TypedMessageData<T>* >(data)->data();
115}
116
117template<class T>
118class DisposeData : public MessageData {
119 public:
120  explicit DisposeData(T* data) : data_(data) { }
121  virtual ~DisposeData() { delete data_; }
122 private:
123  T* data_;
124};
125
126const uint32_t MQID_ANY = static_cast<uint32_t>(-1);
127const uint32_t MQID_DISPOSE = static_cast<uint32_t>(-2);
128
129// No destructor
130
131struct 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
145typedef 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
150class 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
166class 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