/usr/share/cruisecontrol-bin-2.6.1/projects/qpid-trunk/cpp/src/qpid/broker/Queue.h

00001 #ifndef _broker_Queue_h
00002 #define _broker_Queue_h
00003 
00004 /*
00005  *
00006  * Licensed to the Apache Software Foundation (ASF) under one
00007  * or more contributor license agreements.  See the NOTICE file
00008  * distributed with this work for additional information
00009  * regarding copyright ownership.  The ASF licenses this file
00010  * to you under the Apache License, Version 2.0 (the
00011  * "License"); you may not use this file except in compliance
00012  * with the License.  You may obtain a copy of the License at
00013  * 
00014  *   http://www.apache.org/licenses/LICENSE-2.0
00015  * 
00016  * Unless required by applicable law or agreed to in writing,
00017  * software distributed under the License is distributed on an
00018  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
00019  * KIND, either express or implied.  See the License for the
00020  * specific language governing permissions and limitations
00021  * under the License.
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             //PersistableQueue support:
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             // Manageable entry points
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  

Generated on Thu Apr 10 11:08:17 2008 for Qpid by  doxygen 1.4.7