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 OwnershipToken* exclusive;
00074 bool noLocal;
00075 std::string traceId;
00076 std::vector<std::string> traceExclude;
00077 Listeners listeners;
00078 Messages messages;
00079 mutable qpid::sys::Mutex consumerLock;
00080 mutable qpid::sys::Mutex messageLock;
00081 mutable qpid::sys::Mutex ownershipLock;
00082 mutable uint64_t persistenceId;
00083 framing::FieldTable settings;
00084 std::auto_ptr<QueuePolicy> policy;
00085 bool policyExceeded;
00086 QueueBindings bindings;
00087 boost::shared_ptr<Exchange> alternateExchange;
00088 framing::SequenceNumber sequence;
00089 management::Queue::shared_ptr mgmtObject;
00090
00091 void pop();
00092 void push(boost::intrusive_ptr<Message>& msg);
00093 void setPolicy(std::auto_ptr<QueuePolicy> policy);
00094 bool seek(QueuedMessage& msg, Consumer& position);
00095 bool getNextMessage(QueuedMessage& msg, Consumer& c);
00096 bool consumeNextMessage(QueuedMessage& msg, Consumer& c);
00097 bool browseNextMessage(QueuedMessage& msg, Consumer& c);
00098 bool canExcludeUnwanted();
00099
00100 void notify();
00101 void removeListener(Consumer&);
00102 void addListener(Consumer&);
00103
00104 bool isExcluded(boost::intrusive_ptr<Message>& msg);
00105
00106 public:
00107 virtual void notifyDurableIOComplete();
00108 typedef boost::shared_ptr<Queue> shared_ptr;
00109
00110 typedef std::vector<shared_ptr> vector;
00111
00112 Queue(const string& name, bool autodelete = false,
00113 MessageStore* const store = 0,
00114 const OwnershipToken* const owner = 0,
00115 management::Manageable* parent = 0);
00116 ~Queue();
00117
00118 bool dispatch(Consumer&);
00119
00120 void create(const qpid::framing::FieldTable& settings);
00121 void configure(const qpid::framing::FieldTable& settings);
00122 void destroy();
00123 void bound(const string& exchange, const string& key, const qpid::framing::FieldTable& args);
00124 void unbind(ExchangeRegistry& exchanges, Queue::shared_ptr shared_ref);
00125
00126 bool acquire(const QueuedMessage& msg);
00127
00132 void deliver(boost::intrusive_ptr<Message>& msg);
00137 void process(boost::intrusive_ptr<Message>& msg);
00144 void requeue(const QueuedMessage& msg);
00148 void recover(boost::intrusive_ptr<Message>& msg);
00149
00150 void consume(Consumer& c, bool exclusive = false);
00151 void cancel(Consumer& c);
00152
00153 uint32_t purge();
00154 uint32_t getMessageCount() const;
00155 uint32_t getConsumerCount() const;
00156 inline const string& getName() const { return name; }
00157 bool isExclusiveOwner(const OwnershipToken* const o) const;
00158 void releaseExclusiveOwnership();
00159 bool setExclusiveOwner(const OwnershipToken* const o);
00160 bool hasExclusiveConsumer() const;
00161 bool hasExclusiveOwner() const;
00162 inline bool isDurable() const { return store != 0; }
00163 inline const framing::FieldTable& getSettings() const { return settings; }
00164 inline bool isAutoDelete() const { return autodelete; }
00165 bool canAutoDelete() const;
00166
00167 bool enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg);
00171 bool dequeue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg);
00175 QueuedMessage dequeue();
00176
00177 const QueuePolicy* getPolicy();
00178
00179 void setAlternateExchange(boost::shared_ptr<Exchange> exchange);
00180 boost::shared_ptr<Exchange> getAlternateExchange();
00181 bool isLocal(boost::intrusive_ptr<Message>& msg);
00182
00183
00184 uint64_t getPersistenceId() const;
00185 void setPersistenceId(uint64_t persistenceId) const;
00186 void encode(framing::Buffer& buffer) const;
00187 uint32_t encodedSize() const;
00188
00189 static Queue::shared_ptr decode(QueueRegistry& queues, framing::Buffer& buffer);
00190 static void tryAutoDelete(Broker& broker, Queue::shared_ptr);
00191
00192 virtual void setExternalQueueStore(ExternalQueueStore* inst);
00193
00194
00195 management::ManagementObject::shared_ptr GetManagementObject (void) const;
00196 management::Manageable::status_t
00197 ManagementMethod (uint32_t methodId, management::Args& args);
00198 };
00199 }
00200 }
00201
00202
00203 #endif