00001 #ifndef _broker_Queue_h
00002 #define _broker_Queue_h
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024 #include "OwnershipToken.h"
00025 #include "Consumer.h"
00026 #include "Message.h"
00027 #include "PersistableQueue.h"
00028 #include "QueuePolicy.h"
00029 #include "QueueBindings.h"
00030
00031 #include "qpid/framing/FieldTable.h"
00032 #include "qpid/sys/Serializer.h"
00033 #include "qpid/sys/Monitor.h"
00034 #include "qpid/management/Manageable.h"
00035 #include "qpid/management/Queue.h"
00036 #include "qpid/framing/amqp_types.h"
00037
00038 #include <vector>
00039 #include <memory>
00040 #include <deque>
00041 #include <set>
00042
00043 #include <boost/shared_ptr.hpp>
00044 #include <boost/enable_shared_from_this.hpp>
00045 #include <boost/intrusive_ptr.hpp>
00046
00047 namespace qpid {
00048 namespace broker {
00049 class Broker;
00050 class MessageStore;
00051 class QueueRegistry;
00052 class TransactionContext;
00053 class Exchange;
00054
00055 using std::string;
00056
00063 class Queue : public boost::enable_shared_from_this<Queue>,
00064 public PersistableQueue, public management::Manageable {
00065 typedef std::set<Consumer*> Listeners;
00066 typedef std::deque<QueuedMessage> Messages;
00067
00068 const string name;
00069 const bool autodelete;
00070 MessageStore* store;
00071 const OwnershipToken* owner;
00072 uint32_t consumerCount;
00073 bool exclusive;
00074 bool noLocal;
00075 Listeners listeners;
00076 Messages messages;
00077 mutable qpid::sys::Mutex consumerLock;
00078 mutable qpid::sys::Mutex messageLock;
00079 mutable qpid::sys::Mutex ownershipLock;
00080 mutable uint64_t persistenceId;
00081 framing::FieldTable settings;
00082 std::auto_ptr<QueuePolicy> policy;
00083 QueueBindings bindings;
00084 boost::shared_ptr<Exchange> alternateExchange;
00085 framing::SequenceNumber sequence;
00086 management::Queue::shared_ptr mgmtObject;
00087
00088 void pop();
00089 void push(boost::intrusive_ptr<Message>& msg);
00090 void setPolicy(std::auto_ptr<QueuePolicy> policy);
00091 bool seek(QueuedMessage& msg, Consumer& position);
00092 bool getNextMessage(QueuedMessage& msg, Consumer& c);
00093 bool consumeNextMessage(QueuedMessage& msg, Consumer& c);
00094 bool browseNextMessage(QueuedMessage& msg, Consumer& c);
00095 bool canExcludeUnwanted();
00096
00097 void notify();
00098 void removeListener(Consumer&);
00099 void addListener(Consumer&);
00100
00101 public:
00102 virtual void notifyDurableIOComplete();
00103 typedef boost::shared_ptr<Queue> shared_ptr;
00104
00105 typedef std::vector<shared_ptr> vector;
00106
00107 Queue(const string& name, bool autodelete = false,
00108 MessageStore* const store = 0,
00109 const OwnershipToken* const owner = 0,
00110 management::Manageable* parent = 0);
00111 ~Queue();
00112
00113 bool dispatch(Consumer&);
00114
00115 void create(const qpid::framing::FieldTable& settings);
00116 void configure(const qpid::framing::FieldTable& settings);
00117 void destroy();
00118 void bound(const string& exchange, const string& key, const qpid::framing::FieldTable& args);
00119 void unbind(ExchangeRegistry& exchanges, Queue::shared_ptr shared_ref);
00120
00121 bool acquire(const QueuedMessage& msg);
00122
00123 bool isLocal(boost::intrusive_ptr<Message>& msg);
00128 void deliver(boost::intrusive_ptr<Message>& msg);
00133 void process(boost::intrusive_ptr<Message>& msg);
00140 void requeue(const QueuedMessage& msg);
00144 void recover(boost::intrusive_ptr<Message>& msg);
00145
00146 void consume(Consumer& c, bool exclusive = false);
00147 void cancel(Consumer& c);
00148
00149 uint32_t purge();
00150 uint32_t getMessageCount() const;
00151 uint32_t getConsumerCount() const;
00152 inline const string& getName() const { return name; }
00153 bool isExclusiveOwner(const OwnershipToken* const o) const;
00154 void releaseExclusiveOwnership();
00155 bool setExclusiveOwner(const OwnershipToken* const o);
00156 bool hasExclusiveConsumer() const;
00157 bool hasExclusiveOwner() const;
00158 inline bool isDurable() const { return store != 0; }
00159 inline const framing::FieldTable& getSettings() const { return settings; }
00160 inline bool isAutoDelete() const { return autodelete; }
00161 bool canAutoDelete() const;
00162
00163 bool enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg);
00167 bool dequeue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg);
00171 QueuedMessage dequeue();
00172
00173 const QueuePolicy* getPolicy();
00174
00175 void setAlternateExchange(boost::shared_ptr<Exchange> exchange);
00176 boost::shared_ptr<Exchange> getAlternateExchange();
00177
00178
00179 uint64_t getPersistenceId() const;
00180 void setPersistenceId(uint64_t persistenceId) const;
00181 void encode(framing::Buffer& buffer) const;
00182 uint32_t encodedSize() const;
00183
00184 static Queue::shared_ptr decode(QueueRegistry& queues, framing::Buffer& buffer);
00185 static void tryAutoDelete(Broker& broker, Queue::shared_ptr);
00186
00187 virtual void setExternalQueueStore(ExternalQueueStore* inst);
00188
00189
00190 management::ManagementObject::shared_ptr GetManagementObject (void) const;
00191 management::Manageable::status_t
00192 ManagementMethod (uint32_t methodId, management::Args& args);
00193 };
00194 }
00195 }
00196
00197
00198 #endif