You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2012/06/28 11:15:39 UTC
svn commit: r1354874 [6/14] - in /qpid/branches/java-config-and-management:
./ qpid/ qpid/bin/ qpid/cpp/ qpid/cpp/bindings/ qpid/cpp/bindings/qmf/ruby/
qpid/cpp/bindings/qpid/python/ qpid/cpp/bindings/qpid/ruby/
qpid/cpp/bindings/qpid/ruby/features/ qp...
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/PriorityQueue.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/PriorityQueue.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/PriorityQueue.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/PriorityQueue.cpp Thu Jun 28 09:14:52 2012
@@ -3,13 +3,13 @@
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
+ * regarding copyright ownersip. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -22,96 +22,87 @@
#include "qpid/broker/Queue.h"
#include "qpid/broker/QueuedMessage.h"
#include "qpid/framing/reply_exceptions.h"
+#include "qpid/log/Statement.h"
#include <cmath>
namespace qpid {
namespace broker {
-PriorityQueue::PriorityQueue(int l) :
+PriorityQueue::PriorityQueue(int l) :
levels(l),
messages(levels, Deque()),
frontLevel(0), haveFront(false), cached(false) {}
-bool PriorityQueue::deleted(const QueuedMessage&) { return true; }
+bool PriorityQueue::deleted(const QueuedMessage& qm) {
+ bool deleted = fifo.deleted(qm);
+ if (deleted) erase(qm);
+ return deleted;
+}
size_t PriorityQueue::size()
{
- size_t total(0);
- for (int i = 0; i < levels; ++i) {
- total += messages[i].size();
- }
- return total;
+ return fifo.size();
+}
+
+namespace {
+bool before(QueuedMessage* a, QueuedMessage* b) { return *a < *b; }
}
void PriorityQueue::release(const QueuedMessage& message)
{
- uint p = getPriorityLevel(message);
- messages[p].insert(lower_bound(messages[p].begin(), messages[p].end(), message), message);
- clearCache();
+ QueuedMessage* qm = fifo.releasePtr(message);
+ if (qm) {
+ uint p = getPriorityLevel(message);
+ messages[p].insert(
+ lower_bound(messages[p].begin(), messages[p].end(), qm, before), qm);
+ clearCache();
+ }
}
-bool PriorityQueue::find(const framing::SequenceNumber& position, QueuedMessage& message, bool remove)
-{
- QueuedMessage comp;
- comp.position = position;
- for (int i = 0; i < levels; ++i) {
- if (!messages[i].empty()) {
- unsigned long diff = position.getValue() - messages[i].front().position.getValue();
- long maxEnd = diff < messages[i].size() ? diff : messages[i].size();
- Deque::iterator l = lower_bound(messages[i].begin(),messages[i].begin()+maxEnd,comp);
- if (l != messages[i].end() && l->position == position) {
- message = *l;
- if (remove) {
- messages[i].erase(l);
- clearCache();
- }
- return true;
- }
+
+void PriorityQueue::erase(const QueuedMessage& qm) {
+ size_t i = getPriorityLevel(qm);
+ if (!messages[i].empty()) {
+ long diff = qm.position.getValue() - messages[i].front()->position.getValue();
+ if (diff < 0) return;
+ long maxEnd = std::min(size_t(diff), messages[i].size());
+ QueuedMessage mutableQm = qm; // need non-const qm for lower_bound
+ Deque::iterator l =
+ lower_bound(messages[i].begin(),messages[i].begin()+maxEnd, &mutableQm, before);
+ if (l != messages[i].end() && (*l)->position == qm.position) {
+ messages[i].erase(l);
+ clearCache();
+ return;
}
}
- return false;
}
bool PriorityQueue::acquire(const framing::SequenceNumber& position, QueuedMessage& message)
{
- return find(position, message, true);
+ bool acquired = fifo.acquire(position, message);
+ if (acquired) erase(message); // No longer available
+ return acquired;
}
bool PriorityQueue::find(const framing::SequenceNumber& position, QueuedMessage& message)
{
- return find(position, message, false);
+ return fifo.find(position, message);
}
-bool PriorityQueue::browse(const framing::SequenceNumber& position, QueuedMessage& message, bool)
+bool PriorityQueue::browse(
+ const framing::SequenceNumber& position, QueuedMessage& message, bool unacquired)
{
- QueuedMessage match;
- match.position = position+1;
- Deque::iterator lowest;
- bool found = false;
- for (int i = 0; i < levels; ++i) {
- Deque::iterator m = lower_bound(messages[i].begin(), messages[i].end(), match);
- if (m != messages[i].end()) {
- if (m->position == match.position) {
- message = *m;
- return true;
- } else if (!found || m->position < lowest->position) {
- lowest = m;
- found = true;
- }
- }
- }
- if (found) {
- message = *lowest;
- }
- return found;
+ return fifo.browse(position, message, unacquired);
}
bool PriorityQueue::consume(QueuedMessage& message)
{
if (checkFront()) {
- message = messages[frontLevel].front();
+ QueuedMessage* pm = messages[frontLevel].front();
messages[frontLevel].pop_front();
clearCache();
+ pm->status = QueuedMessage::ACQUIRED; // Updates FIFO index
+ message = *pm;
return true;
} else {
return false;
@@ -120,23 +111,31 @@ bool PriorityQueue::consume(QueuedMessag
bool PriorityQueue::push(const QueuedMessage& added, QueuedMessage& /*not needed*/)
{
- messages[getPriorityLevel(added)].push_back(added);
+ QueuedMessage* qmp = fifo.pushPtr(added);
+ messages[getPriorityLevel(added)].push_back(qmp);
clearCache();
- return false;//adding a message never causes one to be removed for deque
+ return false; // Adding a message never causes one to be removed for deque
+}
+
+void PriorityQueue::updateAcquired(const QueuedMessage& acquired) {
+ fifo.updateAcquired(acquired);
+}
+
+void PriorityQueue::setPosition(const framing::SequenceNumber& n) {
+ fifo.setPosition(n);
}
void PriorityQueue::foreach(Functor f)
{
- for (int i = 0; i < levels; ++i) {
- std::for_each(messages[i].begin(), messages[i].end(), f);
- }
+ fifo.foreach(f);
}
void PriorityQueue::removeIf(Predicate p)
{
for (int priority = 0; priority < levels; ++priority) {
for (Deque::iterator i = messages[priority].begin(); i != messages[priority].end();) {
- if (p(*i)) {
+ if (p(**i)) {
+ (*i)->status = QueuedMessage::DELETED; // Updates fifo index
i = messages[priority].erase(i);
clearCache();
} else {
@@ -144,6 +143,7 @@ void PriorityQueue::removeIf(Predicate p
}
}
}
+ fifo.clean();
}
uint PriorityQueue::getPriorityLevel(const QueuedMessage& m) const
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/PriorityQueue.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/PriorityQueue.h?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/PriorityQueue.h (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/PriorityQueue.h Thu Jun 28 09:14:52 2012
@@ -10,9 +10,9 @@
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -21,7 +21,7 @@
* under the License.
*
*/
-#include "qpid/broker/Messages.h"
+#include "qpid/broker/MessageDeque.h"
#include "qpid/sys/IntegerTypes.h"
#include <deque>
#include <vector>
@@ -32,7 +32,10 @@ namespace broker {
/**
* Basic priority queue with a configurable number of recognised
* priority levels. This is implemented as a separate deque per
- * priority level. Browsing is FIFO not priority order.
+ * priority level.
+ *
+ * Browsing is FIFO not priority order. There is a MessageDeque
+ * for fast browsing.
*/
class PriorityQueue : public Messages
{
@@ -48,23 +51,32 @@ class PriorityQueue : public Messages
bool browse(const framing::SequenceNumber&, QueuedMessage&, bool);
bool consume(QueuedMessage&);
bool push(const QueuedMessage& added, QueuedMessage& removed);
-
+ void updateAcquired(const QueuedMessage& acquired);
+ void setPosition(const framing::SequenceNumber&);
void foreach(Functor);
void removeIf(Predicate);
+
static uint getPriority(const QueuedMessage&);
+
protected:
- typedef std::deque<QueuedMessage> Deque;
+ typedef std::deque<QueuedMessage*> Deque;
typedef std::vector<Deque> PriorityLevels;
virtual bool findFrontLevel(uint& p, PriorityLevels&);
const int levels;
+
private:
+ /** Available messages separated by priority and sorted in priority order.
+ * Holds pointers to the QueuedMessages in fifo
+ */
PriorityLevels messages;
+ /** FIFO index of all messsagse (including acquired messages) for fast browsing and indexing */
+ MessageDeque fifo;
uint frontLevel;
bool haveFront;
bool cached;
-
- bool find(const framing::SequenceNumber&, QueuedMessage&, bool remove);
+
+ void erase(const QueuedMessage&);
uint getPriorityLevel(const QueuedMessage&) const;
void clearCache();
bool checkFront();
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/Queue.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/Queue.cpp Thu Jun 28 09:14:52 2012
@@ -19,8 +19,9 @@
*
*/
-#include "qpid/broker/Broker.h"
#include "qpid/broker/Queue.h"
+
+#include "qpid/broker/Broker.h"
#include "qpid/broker/QueueEvents.h"
#include "qpid/broker/Exchange.h"
#include "qpid/broker/Fairshare.h"
@@ -41,12 +42,14 @@
#include "qpid/management/ManagementAgent.h"
#include "qpid/framing/reply_exceptions.h"
#include "qpid/framing/FieldTable.h"
+#include "qpid/framing/FieldValue.h"
#include "qpid/sys/ClusterSafe.h"
#include "qpid/sys/Monitor.h"
#include "qpid/sys/Time.h"
#include "qpid/types/Variant.h"
#include "qmf/org/apache/qpid/broker/ArgsQueuePurge.h"
#include "qmf/org/apache/qpid/broker/ArgsQueueReroute.h"
+#include "qmf/org/apache/qpid/broker/EventQueueDelete.h"
#include <iostream>
#include <algorithm>
@@ -56,13 +59,16 @@
#include <boost/intrusive_ptr.hpp>
-using namespace qpid::broker;
+namespace qpid {
+namespace broker {
+
using namespace qpid::sys;
using namespace qpid::framing;
using qpid::management::ManagementAgent;
using qpid::management::ManagementObject;
using qpid::management::Manageable;
using qpid::management::Args;
+using std::string;
using std::for_each;
using std::mem_fun;
namespace _qmf = qmf::org::apache::qpid::broker;
@@ -150,6 +156,7 @@ Queue::Queue(const string& _name, bool _
store(_store),
owner(_owner),
consumerCount(0),
+ browserCount(0),
exclusive(0),
noLocal(false),
persistLastNode(false),
@@ -232,11 +239,16 @@ void Queue::deliver(boost::intrusive_ptr
void Queue::recoverPrepared(boost::intrusive_ptr<Message>& msg)
{
+ Mutex::ScopedLock locker(messageLock);
if (policy.get()) policy->recoverEnqueued(msg);
}
-void Queue::recover(boost::intrusive_ptr<Message>& msg){
- if (policy.get()) policy->recoverEnqueued(msg);
+void Queue::recover(boost::intrusive_ptr<Message>& msg)
+{
+ {
+ Mutex::ScopedLock locker(messageLock);
+ if (policy.get()) policy->recoverEnqueued(msg);
+ }
push(msg, true);
if (store){
@@ -276,7 +288,6 @@ void Queue::requeue(const QueuedMessage&
assertClusterSafe();
QueueListeners::NotificationSet copy;
{
- Mutex::ScopedLock locker(messageLock);
if (!isEnqueued(msg)) return;
if (deleted) {
//
@@ -294,8 +305,18 @@ void Queue::requeue(const QueuedMessage&
}
mgntDeqStats(msg.payload, mgmtObject, brokerMgmtObject);
} else {
- messages->release(msg);
- listeners.populate(copy);
+ {
+ Mutex::ScopedLock locker(messageLock);
+ messages->release(msg);
+ observeRequeue(msg, locker);
+ listeners.populate(copy);
+ }
+
+ if (mgmtObject) {
+ mgmtObject->inc_releases();
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_releases();
+ }
// for persistLastNode - don't force a message twice to disk, but force it if no force before
if(inLastNodeFailure && persistLastNode && !msg.payload->isStoredOnQueue(shared_from_this())) {
@@ -305,7 +326,6 @@ void Queue::requeue(const QueuedMessage&
enqueue(0, payload);
}
}
- observeRequeue(msg, locker);
}
}
copy.notify();
@@ -313,10 +333,9 @@ void Queue::requeue(const QueuedMessage&
bool Queue::acquireMessageAt(const SequenceNumber& position, QueuedMessage& message)
{
- Mutex::ScopedLock locker(messageLock);
assertClusterSafe();
QPID_LOG(debug, "Attempting to acquire message at " << position);
- if (acquire(position, message, locker)) {
+ if (acquire(position, message)) {
QPID_LOG(debug, "Acquired message at " << position << " from " << name);
return true;
} else {
@@ -327,17 +346,20 @@ bool Queue::acquireMessageAt(const Seque
bool Queue::acquire(const QueuedMessage& msg, const std::string& consumer)
{
- Mutex::ScopedLock locker(messageLock);
assertClusterSafe();
QPID_LOG(debug, consumer << " attempting to acquire message at " << msg.position);
-
- if (!allocator->allocate( consumer, msg )) {
+ bool ok;
+ {
+ Mutex::ScopedLock locker(messageLock);
+ ok = allocator->allocate( consumer, msg );
+ }
+ if (!ok) {
QPID_LOG(debug, "Not permitted to acquire msg at " << msg.position << " from '" << name);
return false;
}
QueuedMessage copy(msg);
- if (acquire( msg.position, copy, locker)) {
+ if (acquire( msg.position, copy)) {
QPID_LOG(debug, "Acquired message at " << msg.position << " from " << name);
return true;
}
@@ -379,59 +401,73 @@ bool Queue::getNextMessage(QueuedMessage
Queue::ConsumeCode Queue::consumeNextMessage(QueuedMessage& m, Consumer::shared_ptr& c)
{
while (true) {
- Mutex::ScopedLock locker(messageLock);
QueuedMessage msg;
- if (allocator->nextConsumableMessage(c, msg)) {
- if (msg.payload->hasExpired()) {
- QPID_LOG(debug, "Message expired from queue '" << name << "'");
- c->setPosition(msg.position);
- dequeue(0, msg);
- if (mgmtObject) {
- mgmtObject->inc_discardsTtl();
- if (brokerMgmtObject)
- brokerMgmtObject->inc_discardsTtl();
- }
+ bool found;
+ {
+ Mutex::ScopedLock locker(messageLock);
+ found = allocator->nextConsumableMessage(c, msg);
+ if (!found) listeners.addListener(c);
+ }
+ if (!found) {
+ QPID_LOG(debug, "No messages to dispatch on queue '" << name << "'");
+ return NO_MESSAGES;
+ }
- continue;
+ if (msg.payload->hasExpired()) {
+ QPID_LOG(debug, "Message expired from queue '" << name << "'");
+ c->setPosition(msg.position);
+ dequeue(0, msg);
+ if (mgmtObject) {
+ mgmtObject->inc_discardsTtl();
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_discardsTtl();
}
+ continue;
+ }
- if (c->filter(msg.payload)) {
- if (c->accept(msg.payload)) {
+ if (c->filter(msg.payload)) {
+ if (c->accept(msg.payload)) {
+ {
+ Mutex::ScopedLock locker(messageLock);
bool ok = allocator->allocate( c->getName(), msg ); // inform allocator
(void) ok; assert(ok);
observeAcquire(msg, locker);
- m = msg;
- return CONSUMED;
- } else {
- //message(s) are available but consumer hasn't got enough credit
- QPID_LOG(debug, "Consumer can't currently accept message from '" << name << "'");
- messages->release(msg);
- return CANT_CONSUME;
}
+ if (mgmtObject) {
+ mgmtObject->inc_acquires();
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_acquires();
+ }
+ m = msg;
+ return CONSUMED;
} else {
- //consumer will never want this message
- QPID_LOG(debug, "Consumer doesn't want message from '" << name << "'");
- messages->release(msg);
- return CANT_CONSUME;
+ //message(s) are available but consumer hasn't got enough credit
+ QPID_LOG(debug, "Consumer can't currently accept message from '" << name << "'");
}
} else {
- QPID_LOG(debug, "No messages to dispatch on queue '" << name << "'");
- listeners.addListener(c);
- return NO_MESSAGES;
+ //consumer will never want this message
+ QPID_LOG(debug, "Consumer doesn't want message from '" << name << "'");
}
+
+ Mutex::ScopedLock locker(messageLock);
+ messages->release(msg);
+ return CANT_CONSUME;
}
}
bool Queue::browseNextMessage(QueuedMessage& m, Consumer::shared_ptr& c)
{
while (true) {
- Mutex::ScopedLock locker(messageLock);
QueuedMessage msg;
-
- if (!allocator->nextBrowsableMessage(c, msg)) { // no next available
+ bool found;
+ {
+ Mutex::ScopedLock locker(messageLock);
+ found = allocator->nextBrowsableMessage(c, msg);
+ if (!found) listeners.addListener(c);
+ }
+ if (!found) { // no next available
QPID_LOG(debug, "No browsable messages available for consumer " <<
c->getName() << " on queue '" << name << "'");
- listeners.addListener(c);
return false;
}
@@ -489,78 +525,115 @@ bool Queue::find(SequenceNumber pos, Que
void Queue::consume(Consumer::shared_ptr c, bool requestExclusive){
assertClusterSafe();
{
- Mutex::ScopedLock locker(consumerLock);
- if(exclusive) {
- throw ResourceLockedException(
- QPID_MSG("Queue " << getName() << " has an exclusive consumer. No more consumers allowed."));
- } else if(requestExclusive) {
- if(consumerCount) {
+ Mutex::ScopedLock locker(messageLock);
+ // NOTE: consumerCount is actually a count of all
+ // subscriptions, both acquiring and non-acquiring (browsers).
+ // Check for exclusivity of acquiring consumers.
+ size_t acquiringConsumers = consumerCount - browserCount;
+ if (c->preAcquires()) {
+ if(exclusive) {
throw ResourceLockedException(
- QPID_MSG("Queue " << getName() << " already has consumers. Exclusive access denied."));
- } else {
- exclusive = c->getSession();
+ QPID_MSG("Queue " << getName()
+ << " has an exclusive consumer. No more consumers allowed."));
+ } else if(requestExclusive) {
+ if(acquiringConsumers) {
+ throw ResourceLockedException(
+ QPID_MSG("Queue " << getName()
+ << " already has consumers. Exclusive access denied."));
+ } else {
+ exclusive = c->getSession();
+ }
}
}
+ else
+ browserCount++;
consumerCount++;
- if (mgmtObject != 0)
- mgmtObject->inc_consumerCount ();
//reset auto deletion timer if necessary
if (autoDeleteTimeout && autoDeleteTask) {
autoDeleteTask->cancel();
}
+ observeConsumerAdd(*c, locker);
}
- Mutex::ScopedLock locker(messageLock);
- for (Observers::const_iterator i = observers.begin(); i != observers.end(); ++i) {
- try{
- (*i)->consumerAdded(*c);
- } catch (const std::exception& e) {
- QPID_LOG(warning, "Exception on notification of new consumer for queue " << getName() << ": " << e.what());
- }
- }
+ if (mgmtObject != 0)
+ mgmtObject->inc_consumerCount ();
}
void Queue::cancel(Consumer::shared_ptr c){
removeListener(c);
{
- Mutex::ScopedLock locker(consumerLock);
+ Mutex::ScopedLock locker(messageLock);
consumerCount--;
+ if (!c->preAcquires()) browserCount--;
if(exclusive) exclusive = 0;
- if (mgmtObject != 0)
- mgmtObject->dec_consumerCount ();
- }
- Mutex::ScopedLock locker(messageLock);
- for (Observers::const_iterator i = observers.begin(); i != observers.end(); ++i) {
- try{
- (*i)->consumerRemoved(*c);
- } catch (const std::exception& e) {
- QPID_LOG(warning, "Exception on notification of removed consumer for queue " << getName() << ": " << e.what());
- }
+ observeConsumerRemove(*c, locker);
}
+ if (mgmtObject != 0)
+ mgmtObject->dec_consumerCount ();
}
QueuedMessage Queue::get(){
- Mutex::ScopedLock locker(messageLock);
QueuedMessage msg(this);
- if (messages->consume(msg))
- observeAcquire(msg, locker);
+ bool ok;
+ {
+ Mutex::ScopedLock locker(messageLock);
+ ok = messages->consume(msg);
+ if (ok) observeAcquire(msg, locker);
+ }
+
+ if (ok && mgmtObject) {
+ mgmtObject->inc_acquires();
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_acquires();
+ }
+
return msg;
}
-bool collect_if_expired(std::deque<QueuedMessage>& expired, QueuedMessage& message)
+namespace {
+bool collectIf(QueuedMessage& qm, Messages::Predicate predicate,
+ std::deque<QueuedMessage>& collection)
{
- if (message.payload->hasExpired()) {
- expired.push_back(message);
+ if (predicate(qm)) {
+ collection.push_back(qm);
return true;
} else {
return false;
}
}
+bool isExpired(const QueuedMessage& qm) { return qm.payload->hasExpired(); }
+} // namespace
+
+void Queue::dequeueIf(Messages::Predicate predicate,
+ std::deque<QueuedMessage>& dequeued)
+{
+ {
+ Mutex::ScopedLock locker(messageLock);
+ messages->removeIf(boost::bind(&collectIf, _1, predicate, boost::ref(dequeued)));
+ }
+ if (!dequeued.empty()) {
+ if (mgmtObject) {
+ mgmtObject->inc_acquires(dequeued.size());
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_acquires(dequeued.size());
+ }
+ for (std::deque<QueuedMessage>::const_iterator i = dequeued.begin();
+ i != dequeued.end(); ++i) {
+ {
+ // KAG: should be safe to retake lock after the removeIf, since
+ // no other thread can touch these messages after the removeIf() call
+ Mutex::ScopedLock locker(messageLock);
+ observeAcquire(*i, locker);
+ }
+ dequeue( 0, *i );
+ }
+ }
+}
+
/**
*@param lapse: time since the last purgeExpired
*/
-void Queue::purgeExpired(qpid::sys::Duration lapse)
-{
+void Queue::purgeExpired(sys::Duration lapse) {
//As expired messages are discarded during dequeue also, only
//bother explicitly expiring if the rate of dequeues since last
//attempt is less than one per second.
@@ -568,33 +641,18 @@ void Queue::purgeExpired(qpid::sys::Dura
dequeueSincePurge -= count;
int seconds = int64_t(lapse)/qpid::sys::TIME_SEC;
if (seconds == 0 || count / seconds < 1) {
- std::deque<QueuedMessage> expired;
- {
- Mutex::ScopedLock locker(messageLock);
- messages->removeIf(boost::bind(&collect_if_expired, boost::ref(expired), _1));
- }
-
- //
- // Report the count of discarded-by-ttl messages
- //
- if (mgmtObject && !expired.empty()) {
- mgmtObject->inc_discardsTtl(expired.size());
- if (brokerMgmtObject)
- brokerMgmtObject->inc_discardsTtl(expired.size());
- }
-
- for (std::deque<QueuedMessage>::const_iterator i = expired.begin();
- i != expired.end(); ++i) {
- {
- Mutex::ScopedLock locker(messageLock);
- observeAcquire(*i, locker);
+ std::deque<QueuedMessage> dequeued;
+ dequeueIf(boost::bind(&isExpired, _1), dequeued);
+ if (dequeued.size()) {
+ if (mgmtObject) {
+ mgmtObject->inc_discardsTtl(dequeued.size());
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_discardsTtl(dequeued.size());
}
- dequeue( 0, *i );
}
}
}
-
namespace {
// for use with purge/move below - collect messages that match a given filter
//
@@ -715,32 +773,47 @@ uint32_t Queue::purge(const uint32_t pur
std::auto_ptr<MessageFilter> mf(MessageFilter::create(filter));
Collector c(*mf.get(), purge_request);
- Mutex::ScopedLock locker(messageLock);
- messages->removeIf( boost::bind<bool>(boost::ref(c), _1) );
+ {
+ Mutex::ScopedLock locker(messageLock);
+ messages->removeIf( boost::bind<bool>(boost::ref(c), _1) );
+ }
- if (mgmtObject && !c.matches.empty()) {
- if (dest.get()) {
- mgmtObject->inc_reroutes(c.matches.size());
- if (brokerMgmtObject)
- brokerMgmtObject->inc_reroutes(c.matches.size());
- } else {
- mgmtObject->inc_discardsPurge(c.matches.size());
- if (brokerMgmtObject)
- brokerMgmtObject->inc_discardsPurge(c.matches.size());
+ if (!c.matches.empty()) {
+ if (mgmtObject) {
+ mgmtObject->inc_acquires(c.matches.size());
+ if (dest.get()) {
+ mgmtObject->inc_reroutes(c.matches.size());
+ if (brokerMgmtObject) {
+ brokerMgmtObject->inc_acquires(c.matches.size());
+ brokerMgmtObject->inc_reroutes(c.matches.size());
+ }
+ } else {
+ mgmtObject->inc_discardsPurge(c.matches.size());
+ if (brokerMgmtObject) {
+ brokerMgmtObject->inc_acquires(c.matches.size());
+ brokerMgmtObject->inc_discardsPurge(c.matches.size());
+ }
+ }
}
- }
- for (std::deque<QueuedMessage>::iterator qmsg = c.matches.begin();
- qmsg != c.matches.end(); ++qmsg) {
- // Update observers and message state:
- observeAcquire(*qmsg, locker);
- dequeue(0, *qmsg);
- QPID_LOG(debug, "Purged message at " << qmsg->position << " from " << getName());
- // now reroute if necessary
- if (dest.get()) {
- assert(qmsg->payload);
- DeliverableMessage dmsg(qmsg->payload);
- dest->routeWithAlternate(dmsg);
+ for (std::deque<QueuedMessage>::iterator qmsg = c.matches.begin();
+ qmsg != c.matches.end(); ++qmsg) {
+
+ {
+ // KAG: should be safe to retake lock after the removeIf, since
+ // no other thread can touch these messages after the removeIf call
+ Mutex::ScopedLock locker(messageLock);
+ observeAcquire(*qmsg, locker);
+ }
+ dequeue(0, *qmsg);
+ QPID_LOG(debug, "Purged message at " << qmsg->position << " from " << getName());
+ // now reroute if necessary
+ if (dest.get()) {
+ assert(qmsg->payload);
+ qmsg->payload->clearTrace();
+ DeliverableMessage dmsg(qmsg->payload);
+ dest->routeWithAlternate(dmsg);
+ }
}
}
return c.matches.size();
@@ -752,27 +825,51 @@ uint32_t Queue::move(const Queue::shared
std::auto_ptr<MessageFilter> mf(MessageFilter::create(filter));
Collector c(*mf.get(), qty);
- Mutex::ScopedLock locker(messageLock);
- messages->removeIf( boost::bind<bool>(boost::ref(c), _1) );
+ {
+ Mutex::ScopedLock locker(messageLock);
+ messages->removeIf( boost::bind<bool>(boost::ref(c), _1) );
+ }
+
- for (std::deque<QueuedMessage>::iterator qmsg = c.matches.begin();
- qmsg != c.matches.end(); ++qmsg) {
+ if (!c.matches.empty()) {
// Update observers and message state:
- observeAcquire(*qmsg, locker);
- dequeue(0, *qmsg);
- // and move to destination Queue.
- assert(qmsg->payload);
- destq->deliver(qmsg->payload);
+
+ if (mgmtObject) {
+ mgmtObject->inc_acquires(c.matches.size());
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_acquires(c.matches.size());
+ }
+
+ for (std::deque<QueuedMessage>::iterator qmsg = c.matches.begin();
+ qmsg != c.matches.end(); ++qmsg) {
+ {
+ Mutex::ScopedLock locker(messageLock);
+ observeAcquire(*qmsg, locker);
+ }
+ dequeue(0, *qmsg);
+ // and move to destination Queue.
+ assert(qmsg->payload);
+ destq->deliver(qmsg->payload);
+ }
}
return c.matches.size();
}
/** Acquire the message at the given position, return true and msg if acquire succeeds */
-bool Queue::acquire(const qpid::framing::SequenceNumber& position, QueuedMessage& msg,
- const Mutex::ScopedLock& locker)
+bool Queue::acquire(const qpid::framing::SequenceNumber& position, QueuedMessage& msg)
{
- if (messages->acquire(position, msg)) {
- observeAcquire(msg, locker);
+ bool ok;
+ {
+ Mutex::ScopedLock locker(messageLock);
+ ok = messages->acquire(position, msg);
+ if (ok) observeAcquire(msg, locker);
+ }
+ if (ok) {
+ if (mgmtObject) {
+ mgmtObject->inc_acquires();
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_acquires();
+ }
++dequeueSincePurge;
return true;
}
@@ -782,35 +879,44 @@ bool Queue::acquire(const qpid::framing:
void Queue::push(boost::intrusive_ptr<Message>& msg, bool isRecovery){
assertClusterSafe();
QueueListeners::NotificationSet copy;
- QueuedMessage removed;
+ QueuedMessage removed, qm(this, msg);
bool dequeueRequired = false;
{
Mutex::ScopedLock locker(messageLock);
- QueuedMessage qm(this, msg, ++sequence);
- if (insertSeqNo) msg->insertCustomProperty(seqNoKey, sequence);
-
- dequeueRequired = messages->push(qm, removed);
- if (dequeueRequired) {
+ qm.position = ++sequence;
+ if (messages->push(qm, removed)) {
+ dequeueRequired = true;
observeAcquire(removed, locker);
- if (mgmtObject) {
- mgmtObject->inc_discardsLvq();
- if (brokerMgmtObject)
- brokerMgmtObject->inc_discardsLvq();
- }
}
- listeners.populate(copy);
observeEnqueue(qm, locker);
+ if (policy.get()) {
+ policy->enqueued(qm);
+ }
+ listeners.populate(copy);
}
- copy.notify();
+ if (insertSeqNo) msg->insertCustomProperty(seqNoKey, qm.position);
+
+ mgntEnqStats(msg, mgmtObject, brokerMgmtObject);
+
if (dequeueRequired) {
+ if (mgmtObject) {
+ mgmtObject->inc_acquires();
+ mgmtObject->inc_discardsLvq();
+ if (brokerMgmtObject) {
+ brokerMgmtObject->inc_acquires();
+ brokerMgmtObject->inc_discardsLvq();
+ }
+ }
if (isRecovery) {
//can't issue new requests for the store until
//recovery is complete
+ Mutex::ScopedLock locker(messageLock);
pendingDequeues.push_back(removed);
} else {
dequeue(0, removed);
}
}
+ copy.notify();
}
void isEnqueueComplete(uint32_t* result, const QueuedMessage& message)
@@ -821,8 +927,8 @@ void isEnqueueComplete(uint32_t* result,
/** function only provided for unit tests, or code not in critical message path */
uint32_t Queue::getEnqueueCompleteMessageCount() const
{
- Mutex::ScopedLock locker(messageLock);
uint32_t count = 0;
+ Mutex::ScopedLock locker(messageLock);
messages->foreach(boost::bind(&isEnqueueComplete, &count, _1));
return count;
}
@@ -835,13 +941,13 @@ uint32_t Queue::getMessageCount() const
uint32_t Queue::getConsumerCount() const
{
- Mutex::ScopedLock locker(consumerLock);
+ Mutex::ScopedLock locker(messageLock);
return consumerCount;
}
bool Queue::canAutoDelete() const
{
- Mutex::ScopedLock locker(consumerLock);
+ Mutex::ScopedLock locker(messageLock);
return autodelete && !consumerCount && !owner;
}
@@ -948,14 +1054,20 @@ bool Queue::dequeue(TransactionContext*
{
ScopedUse u(barrier);
if (!u.acquired) return false;
-
{
Mutex::ScopedLock locker(messageLock);
if (!isEnqueued(msg)) return false;
if (!ctxt) {
+ if (policy.get()) policy->dequeued(msg);
+ messages->deleted(msg);
observeDequeue(msg, locker);
}
}
+
+ if (!ctxt) {
+ mgntDeqStats(msg.payload, mgmtObject, brokerMgmtObject);
+ }
+
// This check prevents messages which have been forced persistent on one queue from dequeuing
// from another on which no forcing has taken place and thus causing a store error.
bool fp = msg.payload->isForcedPersistent();
@@ -972,8 +1084,13 @@ bool Queue::dequeue(TransactionContext*
void Queue::dequeueCommitted(const QueuedMessage& msg)
{
- Mutex::ScopedLock locker(messageLock);
- observeDequeue(msg, locker);
+ {
+ Mutex::ScopedLock locker(messageLock);
+ if (policy.get()) policy->dequeued(msg);
+ messages->deleted(msg);
+ observeDequeue(msg, locker);
+ }
+ mgntDeqStats(msg.payload, mgmtObject, brokerMgmtObject);
if (mgmtObject != 0) {
_qmf::Queue::PerThreadStats *qStats = mgmtObject->getStatistics();
const uint64_t contentSize = msg.payload->contentSize();
@@ -993,10 +1110,20 @@ void Queue::dequeueCommitted(const Queue
* Removes the first (oldest) message from the in-memory delivery queue as well dequeing
* it from the logical (and persistent if applicable) queue
*/
-bool Queue::popAndDequeue(QueuedMessage& msg, const Mutex::ScopedLock& locker)
+bool Queue::popAndDequeue(QueuedMessage& msg)
{
- if (messages->consume(msg)) {
- observeAcquire(msg, locker);
+ bool popped;
+ {
+ Mutex::ScopedLock locker(messageLock);
+ popped = messages->consume(msg);
+ if (popped) observeAcquire(msg, locker);
+ }
+ if (popped) {
+ if (mgmtObject) {
+ mgmtObject->inc_acquires();
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_acquires();
+ }
dequeue(0, msg);
return true;
} else {
@@ -1006,13 +1133,10 @@ bool Queue::popAndDequeue(QueuedMessage&
/**
* Updates policy and management when a message has been dequeued,
- * expects messageLock to be held
+ * Requires messageLock be held by caller.
*/
-void Queue::observeDequeue(const QueuedMessage& msg, const Mutex::ScopedLock&)
+void Queue::observeDequeue(const QueuedMessage& msg, const qpid::sys::Mutex::ScopedLock&)
{
- mgntDeqStats(msg.payload, mgmtObject, brokerMgmtObject);
- if (policy.get()) policy->dequeued(msg);
- messages->deleted(msg);
for (Observers::const_iterator i = observers.begin(); i != observers.end(); ++i) {
try{
(*i)->dequeued(msg);
@@ -1022,17 +1146,11 @@ void Queue::observeDequeue(const QueuedM
}
}
-/** updates queue observers when a message has become unavailable for transfer,
- * expects messageLock to be held
+/** updates queue observers when a message has become unavailable for transfer.
+ * Requires messageLock be held by caller.
*/
-void Queue::observeAcquire(const QueuedMessage& msg, const Mutex::ScopedLock&)
+void Queue::observeAcquire(const QueuedMessage& msg, const qpid::sys::Mutex::ScopedLock&)
{
- if (mgmtObject) {
- mgmtObject->inc_acquires();
- if (brokerMgmtObject)
- brokerMgmtObject->inc_acquires();
- }
-
for (Observers::const_iterator i = observers.begin(); i != observers.end(); ++i) {
try{
(*i)->acquired(msg);
@@ -1042,17 +1160,11 @@ void Queue::observeAcquire(const QueuedM
}
}
-/** updates queue observers when a message has become re-available for transfer,
- * expects messageLock to be held
+/** updates queue observers when a message has become re-available for transfer
+ * Requires messageLock be held by caller.
*/
-void Queue::observeRequeue(const QueuedMessage& msg, const Mutex::ScopedLock&)
+void Queue::observeRequeue(const QueuedMessage& msg, const qpid::sys::Mutex::ScopedLock&)
{
- if (mgmtObject) {
- mgmtObject->inc_releases();
- if (brokerMgmtObject)
- brokerMgmtObject->inc_releases();
- }
-
for (Observers::const_iterator i = observers.begin(); i != observers.end(); ++i) {
try{
(*i)->requeued(msg);
@@ -1062,6 +1174,33 @@ void Queue::observeRequeue(const QueuedM
}
}
+/** updates queue observers when a new consumer has subscribed to this queue.
+ */
+void Queue::observeConsumerAdd( const Consumer& c, const qpid::sys::Mutex::ScopedLock&)
+{
+ for (Observers::const_iterator i = observers.begin(); i != observers.end(); ++i) {
+ try{
+ (*i)->consumerAdded(c);
+ } catch (const std::exception& e) {
+ QPID_LOG(warning, "Exception on notification of new consumer for queue " << getName() << ": " << e.what());
+ }
+ }
+}
+
+/** updates queue observers when a consumer has unsubscribed from this queue.
+ */
+void Queue::observeConsumerRemove( const Consumer& c, const qpid::sys::Mutex::ScopedLock&)
+{
+ for (Observers::const_iterator i = observers.begin(); i != observers.end(); ++i) {
+ try{
+ (*i)->consumerRemoved(c);
+ } catch (const std::exception& e) {
+ QPID_LOG(warning, "Exception on notification of removed consumer for queue " << getName() << ": " << e.what());
+ }
+ }
+}
+
+
void Queue::create(const FieldTable& _settings)
{
settings = _settings;
@@ -1209,23 +1348,21 @@ void Queue::configureImpl(const FieldTab
void Queue::destroyed()
{
unbind(broker->getExchanges());
- {
- Mutex::ScopedLock locker(messageLock);
- QueuedMessage m;
- while(popAndDequeue(m, locker)) {
- DeliverableMessage msg(m.payload);
- if (alternateExchange.get()) {
- if (brokerMgmtObject)
- brokerMgmtObject->inc_abandonedViaAlt();
- alternateExchange->routeWithAlternate(msg);
- } else {
- if (brokerMgmtObject)
- brokerMgmtObject->inc_abandoned();
- }
+
+ QueuedMessage m;
+ while(popAndDequeue(m)) {
+ DeliverableMessage msg(m.payload);
+ if (alternateExchange.get()) {
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_abandonedViaAlt();
+ alternateExchange->routeWithAlternate(msg);
+ } else {
+ if (brokerMgmtObject)
+ brokerMgmtObject->inc_abandoned();
}
- if (alternateExchange.get())
- alternateExchange->decAlternateUsers();
}
+ if (alternateExchange.get())
+ alternateExchange->decAlternateUsers();
if (store) {
barrier.destroy();
@@ -1236,7 +1373,7 @@ void Queue::destroyed()
if (autoDeleteTask) autoDeleteTask = boost::intrusive_ptr<TimerTask>();
notifyDeleted();
{
- Mutex::ScopedLock locker(messageLock);
+ Mutex::ScopedLock lock(messageLock);
observers.clear();
}
}
@@ -1246,8 +1383,8 @@ void Queue::notifyDeleted()
QueueListeners::ListenerSet set;
{
Mutex::ScopedLock locker(messageLock);
- listeners.snapshot(set);
deleted = true;
+ listeners.snapshot(set);
}
set.notifyAll();
}
@@ -1265,6 +1402,7 @@ void Queue::unbind(ExchangeRegistry& exc
void Queue::setPolicy(std::auto_ptr<QueuePolicy> _policy)
{
+ Mutex::ScopedLock locker(messageLock);
policy = _policy;
if (policy.get())
policy->setQueue(this);
@@ -1272,6 +1410,7 @@ void Queue::setPolicy(std::auto_ptr<Queu
const QueuePolicy* Queue::getPolicy()
{
+ Mutex::ScopedLock locker(messageLock);
return policy.get();
}
@@ -1346,12 +1485,15 @@ boost::shared_ptr<Exchange> Queue::getAl
return alternateExchange;
}
-void tryAutoDeleteImpl(Broker& broker, Queue::shared_ptr queue)
+void tryAutoDeleteImpl(Broker& broker, Queue::shared_ptr queue, const std::string& connectionId, const std::string& userId)
{
if (broker.getQueues().destroyIf(queue->getName(),
boost::bind(boost::mem_fn(&Queue::canAutoDelete), queue))) {
QPID_LOG(debug, "Auto-deleting " << queue->getName());
queue->destroyed();
+
+ if (broker.getManagementAgent())
+ broker.getManagementAgent()->raiseEvent(_qmf::EventQueueDelete(connectionId, userId, queue->getName()));
}
}
@@ -1359,9 +1501,11 @@ struct AutoDeleteTask : qpid::sys::Timer
{
Broker& broker;
Queue::shared_ptr queue;
+ std::string connectionId;
+ std::string userId;
- AutoDeleteTask(Broker& b, Queue::shared_ptr q, AbsTime fireTime)
- : qpid::sys::TimerTask(fireTime, "DelayedAutoDeletion"), broker(b), queue(q) {}
+ AutoDeleteTask(Broker& b, Queue::shared_ptr q, const std::string& cId, const std::string& uId, AbsTime fireTime)
+ : qpid::sys::TimerTask(fireTime, "DelayedAutoDeletion:"+q->getName()), broker(b), queue(q), connectionId(cId), userId(uId) {}
void fire()
{
@@ -1369,19 +1513,19 @@ struct AutoDeleteTask : qpid::sys::Timer
//created, but then became unused again before the task fired;
//in this case ignore this request as there will have already
//been a later task added
- tryAutoDeleteImpl(broker, queue);
+ tryAutoDeleteImpl(broker, queue, connectionId, userId);
}
};
-void Queue::tryAutoDelete(Broker& broker, Queue::shared_ptr queue)
+void Queue::tryAutoDelete(Broker& broker, Queue::shared_ptr queue, const std::string& connectionId, const std::string& userId)
{
if (queue->autoDeleteTimeout && queue->canAutoDelete()) {
AbsTime time(now(), Duration(queue->autoDeleteTimeout * TIME_SEC));
- queue->autoDeleteTask = boost::intrusive_ptr<qpid::sys::TimerTask>(new AutoDeleteTask(broker, queue, time));
+ queue->autoDeleteTask = boost::intrusive_ptr<qpid::sys::TimerTask>(new AutoDeleteTask(broker, queue, connectionId, userId, time));
broker.getClusterTimer().add(queue->autoDeleteTask);
QPID_LOG(debug, "Timed auto-delete for " << queue->getName() << " initiated");
} else {
- tryAutoDeleteImpl(broker, queue);
+ tryAutoDeleteImpl(broker, queue, connectionId, userId);
}
}
@@ -1501,9 +1645,14 @@ Manageable::status_t Queue::ManagementMe
{
_qmf::ArgsQueueReroute& rerouteArgs = (_qmf::ArgsQueueReroute&) args;
boost::shared_ptr<Exchange> dest;
- if (rerouteArgs.i_useAltExchange)
+ if (rerouteArgs.i_useAltExchange) {
+ if (!alternateExchange) {
+ status = Manageable::STATUS_PARAMETER_INVALID;
+ etext = "No alternate-exchange defined";
+ break;
+ }
dest = alternateExchange;
- else {
+ } else {
try {
dest = broker->getExchanges().get(rerouteArgs.i_exchange);
} catch(const std::exception&) {
@@ -1530,13 +1679,28 @@ void Queue::query(qpid::types::Variant::
if (allocator) allocator->query(results);
}
+namespace {
+struct After {
+ framing::SequenceNumber seq;
+ After(framing::SequenceNumber s) : seq(s) {}
+ bool operator()(const QueuedMessage& qm) { return qm.position > seq; }
+};
+} // namespace
+
+
void Queue::setPosition(SequenceNumber n) {
Mutex::ScopedLock locker(messageLock);
+ if (n < sequence) {
+ std::deque<QueuedMessage> dequeued;
+ dequeueIf(After(n), dequeued);
+ messages->setPosition(n);
+ }
sequence = n;
QPID_LOG(trace, "Set position to " << sequence << " on " << getName());
}
SequenceNumber Queue::getPosition() {
+ Mutex::ScopedLock locker(messageLock);
return sequence;
}
@@ -1553,8 +1717,12 @@ void Queue::recoveryComplete(ExchangeReg
<< "\": exchange does not exist.");
}
//process any pending dequeues
- for_each(pendingDequeues.begin(), pendingDequeues.end(), boost::bind(&Queue::dequeue, this, (TransactionContext*) 0, _1));
- pendingDequeues.clear();
+ std::deque<QueuedMessage> pd;
+ {
+ Mutex::ScopedLock locker(messageLock);
+ pendingDequeues.swap(pd);
+ }
+ for_each(pd.begin(), pd.end(), boost::bind(&Queue::dequeue, this, (TransactionContext*) 0, _1));
}
void Queue::insertSequenceNumbers(const std::string& key)
@@ -1564,10 +1732,10 @@ void Queue::insertSequenceNumbers(const
QPID_LOG(debug, "Inserting sequence numbers as " << key);
}
-/** updates queue observers and state when a message has become available for transfer,
- * expects messageLock to be held
+/** updates queue observers and state when a message has become available for transfer
+ * Requires messageLock be held by caller.
*/
-void Queue::observeEnqueue(const QueuedMessage& m, const Mutex::ScopedLock&)
+void Queue::observeEnqueue(const QueuedMessage& m, const qpid::sys::Mutex::ScopedLock&)
{
for (Observers::iterator i = observers.begin(); i != observers.end(); ++i) {
try {
@@ -1576,10 +1744,6 @@ void Queue::observeEnqueue(const QueuedM
QPID_LOG(warning, "Exception on notification of enqueue for queue " << getName() << ": " << e.what());
}
}
- if (policy.get()) {
- policy->enqueued(m);
- }
- mgntEnqStats(m.payload, mgmtObject, brokerMgmtObject);
}
void Queue::updateEnqueued(const QueuedMessage& m)
@@ -1587,12 +1751,16 @@ void Queue::updateEnqueued(const QueuedM
if (m.payload) {
boost::intrusive_ptr<Message> payload = m.payload;
enqueue(0, payload, true);
- messages->updateAcquired(m);
- if (policy.get()) {
- policy->recoverEnqueued(payload);
+ {
+ Mutex::ScopedLock locker(messageLock);
+ messages->updateAcquired(m);
+ observeEnqueue(m, locker);
+ if (policy.get()) {
+ policy->recoverEnqueued(payload);
+ policy->enqueued(m);
+ }
}
- Mutex::ScopedLock locker(messageLock);
- observeEnqueue(m, locker);
+ mgntEnqStats(m.payload, mgmtObject, brokerMgmtObject);
} else {
QPID_LOG(warning, "Queue informed of enqueued message that has no payload");
}
@@ -1600,10 +1768,16 @@ void Queue::updateEnqueued(const QueuedM
bool Queue::isEnqueued(const QueuedMessage& msg)
{
+ Mutex::ScopedLock locker(messageLock);
return !policy.get() || policy->isEnqueued(msg);
}
+// Note: accessing listeners outside of lock is dangerous. Caller must ensure the queue's
+// state is not changed while listeners is referenced.
QueueListeners& Queue::getListeners() { return listeners; }
+
+// Note: accessing messages outside of lock is dangerous. Caller must ensure the queue's
+// state is not changed while messages is referenced.
Messages& Queue::getMessages() { return *messages; }
const Messages& Queue::getMessages() const { return *messages; }
@@ -1616,13 +1790,13 @@ void Queue::checkNotDeleted(const Consum
void Queue::addObserver(boost::shared_ptr<QueueObserver> observer)
{
- Mutex::ScopedLock locker(messageLock);
+ Mutex::ScopedLock lock(messageLock);
observers.insert(observer);
}
void Queue::removeObserver(boost::shared_ptr<QueueObserver> observer)
{
- Mutex::ScopedLock locker(messageLock);
+ Mutex::ScopedLock lock(messageLock);
observers.erase(observer);
}
@@ -1685,7 +1859,7 @@ Queue::UsageBarrier::UsageBarrier(Queue&
bool Queue::UsageBarrier::acquire()
{
- Monitor::ScopedLock l(parent.messageLock);
+ Monitor::ScopedLock l(parent.messageLock); /** @todo: use a dedicated lock instead of messageLock */
if (parent.deleted) {
return false;
} else {
@@ -1706,3 +1880,6 @@ void Queue::UsageBarrier::destroy()
parent.deleted = true;
while (count) parent.messageLock.wait();
}
+
+}}
+
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/Queue.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/Queue.h?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/Queue.h Thu Jun 28 09:14:52 2012
@@ -97,7 +97,8 @@ class Queue : public boost::enable_share
const bool autodelete;
MessageStore* store;
const OwnershipToken* owner;
- uint32_t consumerCount;
+ uint32_t consumerCount; // Actually a count of all subscriptions, acquiring or not.
+ uint32_t browserCount; // Count of non-acquiring subscriptions.
OwnershipToken* exclusive;
bool noLocal;
bool persistLastNode;
@@ -107,7 +108,22 @@ class Queue : public boost::enable_share
QueueListeners listeners;
std::auto_ptr<Messages> messages;
std::deque<QueuedMessage> pendingDequeues;//used to avoid dequeuing during recovery
- mutable qpid::sys::Mutex consumerLock;
+ /** messageLock is used to keep the Queue's state consistent while processing message
+ * events, such as message dispatch, enqueue, acquire, and dequeue. It must be held
+ * while updating certain members in order to keep these members consistent with
+ * each other:
+ * o messages
+ * o sequence
+ * o policy
+ * o listeners
+ * o allocator
+ * o observeXXX() methods
+ * o observers
+ * o pendingDequeues (TBD: move under separate lock)
+ * o exclusive OwnershipToken (TBD: move under separate lock)
+ * o consumerCount (TBD: move under separate lock)
+ * o Queue::UsageBarrier (TBD: move under separate lock)
+ */
mutable qpid::sys::Monitor messageLock;
mutable qpid::sys::Mutex ownershipLock;
mutable uint64_t persistenceId;
@@ -143,22 +159,23 @@ class Queue : public boost::enable_share
bool isExcluded(boost::intrusive_ptr<Message>& msg);
- /** update queue observers, stats, policy, etc when the messages' state changes. Lock
- * must be held by caller */
+ /** update queue observers, stats, policy, etc when the messages' state changes.
+ * messageLock is held by caller */
void observeEnqueue(const QueuedMessage& msg, const sys::Mutex::ScopedLock& lock);
void observeAcquire(const QueuedMessage& msg, const sys::Mutex::ScopedLock& lock);
void observeRequeue(const QueuedMessage& msg, const sys::Mutex::ScopedLock& lock);
void observeDequeue(const QueuedMessage& msg, const sys::Mutex::ScopedLock& lock);
- bool popAndDequeue(QueuedMessage&, const sys::Mutex::ScopedLock& lock);
- // acquire message @ position, return true and set msg if acquire succeeds
- bool acquire(const qpid::framing::SequenceNumber& position, QueuedMessage& msg,
- const sys::Mutex::ScopedLock& held);
+ void observeConsumerAdd( const Consumer&, const sys::Mutex::ScopedLock& lock);
+ void observeConsumerRemove( const Consumer&, const sys::Mutex::ScopedLock& lock);
+ bool popAndDequeue(QueuedMessage&);
+ bool acquire(const qpid::framing::SequenceNumber& position, QueuedMessage& msg);
void forcePersistent(QueuedMessage& msg);
int getEventMode();
void configureImpl(const qpid::framing::FieldTable& settings);
void checkNotDeleted(const Consumer::shared_ptr& c);
void notifyDeleted();
+ void dequeueIf(Messages::Predicate predicate, std::deque<QueuedMessage>& dequeued);
public:
@@ -327,7 +344,7 @@ class Queue : public boost::enable_share
* exclusive owner
*/
static Queue::shared_ptr restore(QueueRegistry& queues, framing::Buffer& buffer);
- static void tryAutoDelete(Broker& broker, Queue::shared_ptr);
+ static void tryAutoDelete(Broker& broker, Queue::shared_ptr, const std::string& connectionId, const std::string& userId);
virtual void setExternalQueueStore(ExternalQueueStore* inst);
@@ -355,15 +372,25 @@ class Queue : public boost::enable_share
/** Apply f to each Observer on the queue */
template <class F> void eachObserver(F f) {
+ sys::Mutex::ScopedLock l(messageLock);
std::for_each<Observers::iterator, F>(observers.begin(), observers.end(), f);
}
- /** Set the position sequence number for the next message on the queue.
- * Must be >= the current sequence number.
- * Used by cluster to replicate queues.
+ /**
+ * Set the sequence number for the back of the queue, the
+ * next message enqueued will be pos+1.
+ * If pos > getPosition() this creates a gap in the sequence numbers.
+ * if pos < getPosition() the back of the queue is reset to pos,
+ *
+ * The _caller_ must ensure that any messages after pos have been dequeued.
+ *
+ * Used by HA/cluster code for queue replication.
*/
QPID_BROKER_EXTERN void setPosition(framing::SequenceNumber pos);
- /** return current position sequence number for the next message on the queue.
+
+ /**
+ *@return sequence number for the back of the queue. The next message pushed
+ * will be at getPosition+1
*/
QPID_BROKER_EXTERN framing::SequenceNumber getPosition();
QPID_BROKER_EXTERN void addObserver(boost::shared_ptr<QueueObserver>);
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp Thu Jun 28 09:14:52 2012
@@ -75,8 +75,8 @@ namespace {
result = v->get<int64_t>();
QPID_LOG(debug, "Got integer value for " << key << ": " << result);
if (result >= 0) return result;
- } else if (v->convertsTo<string>()) {
- string s(v->get<string>());
+ } else if (v->convertsTo<std::string>()) {
+ std::string s(v->get<std::string>());
QPID_LOG(debug, "Got string value for " << key << ": " << s);
std::istringstream convert(s);
if (convert >> result && result >= 0) return result;
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueListeners.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueListeners.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueListeners.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueListeners.cpp Thu Jun 28 09:14:52 2012
@@ -79,10 +79,6 @@ void QueueListeners::NotificationSet::no
std::for_each(browsers.begin(), browsers.end(), boost::mem_fn(&Consumer::notify));
}
-bool QueueListeners::contains(Consumer::shared_ptr c) const {
- return c->inListeners;
-}
-
void QueueListeners::ListenerSet::notifyAll()
{
std::for_each(listeners.begin(), listeners.end(), boost::mem_fn(&Consumer::notify));
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueListeners.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueListeners.h?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueListeners.h (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueListeners.h Thu Jun 28 09:14:52 2012
@@ -30,7 +30,7 @@ namespace broker {
/**
* Track and notify components that wish to be notified of messages
* that become available on a queue.
- *
+ *
* None of the methods defined here are protected by locking. However
* the populate method allows a 'snapshot' to be taken of the
* listeners to be notified. NotificationSet::notify() may then be
@@ -61,11 +61,10 @@ class QueueListeners
friend class QueueListeners;
};
- void addListener(Consumer::shared_ptr);
- void removeListener(Consumer::shared_ptr);
+ void addListener(Consumer::shared_ptr);
+ void removeListener(Consumer::shared_ptr);
void populate(NotificationSet&);
void snapshot(ListenerSet&);
- bool contains(Consumer::shared_ptr c) const;
void notifyAll();
template <class F> void eachListener(F f) {
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueuePolicy.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueuePolicy.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueuePolicy.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueuePolicy.cpp Thu Jun 28 09:14:52 2012
@@ -133,8 +133,8 @@ T getCapacity(const FieldTable& settings
result = v->get<T>();
QPID_LOG(debug, "Got integer value for " << key << ": " << result);
if (result >= 0) return result;
- } else if (v->convertsTo<string>()) {
- string s(v->get<string>());
+ } else if (v->convertsTo<std::string>()) {
+ std::string s(v->get<std::string>());
QPID_LOG(debug, "Got string value for " << key << ": " << s);
std::istringstream convert(s);
if (convert >> result && result >= 0 && convert.eof()) return result;
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueRegistry.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueRegistry.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueueRegistry.cpp Thu Jun 28 09:14:52 2012
@@ -18,6 +18,7 @@
* under the License.
*
*/
+#include "qpid/broker/Broker.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/QueueRegistry.h"
#include "qpid/broker/QueueEvents.h"
@@ -46,35 +47,45 @@ QueueRegistry::declare(const string& dec
definition from persistente
record*/)
{
- RWlock::ScopedWlock locker(lock);
- string name = declareName.empty() ? generateName() : declareName;
- assert(!name.empty());
- QueueMap::iterator i = queues.find(name);
-
- if (i == queues.end()) {
- Queue::shared_ptr queue(new Queue(name, autoDelete, durable ? store : 0, owner, parent, broker));
- if (alternate) {
- queue->setAlternateExchange(alternate);//need to do this *before* create
- alternate->incAlternateUsers();
- }
- if (!recovering) {
- //apply settings & create persistent record if required
- queue->create(arguments);
+ Queue::shared_ptr queue;
+ std::pair<Queue::shared_ptr, bool> result;
+ {
+ RWlock::ScopedWlock locker(lock);
+ string name = declareName.empty() ? generateName() : declareName;
+ assert(!name.empty());
+ QueueMap::iterator i = queues.find(name);
+
+ if (i == queues.end()) {
+ queue.reset(new Queue(name, autoDelete, durable ? store : 0, owner, parent, broker));
+ if (alternate) {
+ queue->setAlternateExchange(alternate);//need to do this *before* create
+ alternate->incAlternateUsers();
+ }
+ if (!recovering) {
+ //apply settings & create persistent record if required
+ queue->create(arguments);
+ } else {
+ //i.e. recovering a queue for which we already have a persistent record
+ queue->configure(arguments);
+ }
+ queues[name] = queue;
+ if (lastNode) queue->setLastNodeFailure();
+ result = std::pair<Queue::shared_ptr, bool>(queue, true);
} else {
- //i.e. recovering a queue for which we already have a persistent record
- queue->configure(arguments);
+ result = std::pair<Queue::shared_ptr, bool>(i->second, false);
}
- queues[name] = queue;
- if (lastNode) queue->setLastNodeFailure();
-
- return std::pair<Queue::shared_ptr, bool>(queue, true);
- } else {
- return std::pair<Queue::shared_ptr, bool>(i->second, false);
}
+ if (broker && queue) broker->getConfigurationObservers().queueCreate(queue);
+ return result;
}
-void QueueRegistry::destroyLH (const string& name){
- queues.erase(name);
+void QueueRegistry::destroyLH (const string& name) {
+ QueueMap::iterator i = queues.find(name);
+ if (i != queues.end()) {
+ Queue::shared_ptr q = i->second;
+ queues.erase(i);
+ if (broker) broker->getConfigurationObservers().queueDestroy(q);
+ }
}
void QueueRegistry::destroy (const string& name){
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueuedMessage.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueuedMessage.h?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueuedMessage.h (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/QueuedMessage.h Thu Jun 28 09:14:52 2012
@@ -22,6 +22,8 @@
#define _QueuedMessage_
#include "qpid/broker/Message.h"
+#include "BrokerImportExport.h"
+#include <iosfwd>
namespace qpid {
namespace broker {
@@ -47,6 +49,7 @@ inline bool operator<(const QueuedMessag
return a.position < b.position;
}
+QPID_BROKER_EXTERN std::ostream& operator<<(std::ostream&, const QueuedMessage&);
}}
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp Thu Jun 28 09:14:52 2012
@@ -144,11 +144,13 @@ RecoverableTransaction::shared_ptr Recov
RecoverableConfig::shared_ptr RecoveryManagerImpl::recoverConfig(framing::Buffer& buffer)
{
string kind;
-
+ uint32_t p = buffer.getPosition();
buffer.getShortString (kind);
- if (kind == "link")
+ buffer.setPosition(p);
+
+ if (Link::isEncodedLink(kind))
return RecoverableConfig::shared_ptr(new RecoverableConfigImpl(Link::decode (links, buffer)));
- else if (kind == "bridge")
+ else if (Bridge::isEncodedBridge(kind))
return RecoverableConfig::shared_ptr(new RecoverableConfigImpl(Bridge::decode (links, buffer)));
return RecoverableConfig::shared_ptr(); // TODO: raise an exception instead
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SaslAuthenticator.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SaslAuthenticator.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SaslAuthenticator.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SaslAuthenticator.cpp Thu Jun 28 09:14:52 2012
@@ -7,9 +7,9 @@
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -23,9 +23,11 @@
# include "config.h"
#endif
+#include "qpid/broker/AclModule.h"
#include "qpid/broker/Connection.h"
#include "qpid/log/Statement.h"
#include "qpid/framing/reply_exceptions.h"
+#include "qpid/framing/FieldValue.h"
#include "qpid/sys/SecuritySettings.h"
#include <boost/format.hpp>
@@ -36,6 +38,7 @@
using qpid::sys::cyrus::CyrusSecurityLayer;
#endif
+using std::string;
using namespace qpid::framing;
using qpid::sys::SecurityLayer;
using qpid::sys::SecuritySettings;
@@ -163,13 +166,17 @@ void SaslAuthenticator::fini(void)
#endif
-std::auto_ptr<SaslAuthenticator> SaslAuthenticator::createAuthenticator(Connection& c, bool isShadow )
+std::auto_ptr<SaslAuthenticator> SaslAuthenticator::createAuthenticator(Connection& c )
{
if (c.getBroker().getOptions().auth) {
- if ( isShadow )
- return std::auto_ptr<SaslAuthenticator>(new NullAuthenticator(c, c.getBroker().getOptions().requireEncrypted));
- else
- return std::auto_ptr<SaslAuthenticator>(new CyrusAuthenticator(c, c.getBroker().getOptions().requireEncrypted));
+ // The cluster creates non-authenticated connections for internal shadow connections
+ // that are never connected to an external client.
+ if ( !c.isAuthenticated() )
+ return std::auto_ptr<SaslAuthenticator>(
+ new NullAuthenticator(c, c.getBroker().getOptions().requireEncrypted));
+ else
+ return std::auto_ptr<SaslAuthenticator>(
+ new CyrusAuthenticator(c, c.getBroker().getOptions().requireEncrypted));
} else {
QPID_LOG(debug, "SASL: No Authentication Performed");
return std::auto_ptr<SaslAuthenticator>(new NullAuthenticator(c, c.getBroker().getOptions().requireEncrypted));
@@ -177,7 +184,7 @@ std::auto_ptr<SaslAuthenticator> SaslAut
}
-NullAuthenticator::NullAuthenticator(Connection& c, bool e) : connection(c), client(c.getOutput()),
+NullAuthenticator::NullAuthenticator(Connection& c, bool e) : connection(c), client(c.getOutput()),
realm(c.getBroker().getOptions().realm), encrypt(e) {}
NullAuthenticator::~NullAuthenticator() {}
@@ -213,7 +220,7 @@ void NullAuthenticator::start(const stri
} else if (i != string::npos) {
//authorization id is first null delimited field
uid = response->substr(0, i);
- }//else not a valid SASL PLAIN response, throw error?
+ }//else not a valid SASL PLAIN response, throw error?
if (!uid.empty()) {
//append realm if it has not already been added
i = uid.find(realm);
@@ -225,7 +232,12 @@ void NullAuthenticator::start(const stri
}
} else {
connection.setUserId("anonymous");
- }
+ }
+ AclModule* acl = connection.getBroker().getAcl();
+ if (acl && !acl->approveConnection(connection))
+ {
+ throw ConnectionForcedException("User connection denied by configured limit");
+ }
client.tune(framing::CHANNEL_MAX, connection.getFrameMax(), 0, connection.getHeartbeatMax());
}
@@ -239,7 +251,7 @@ std::auto_ptr<SecurityLayer> NullAuthent
#if HAVE_SASL
-CyrusAuthenticator::CyrusAuthenticator(Connection& c, bool _encrypt) :
+CyrusAuthenticator::CyrusAuthenticator(Connection& c, bool _encrypt) :
sasl_conn(0), connection(c), client(c.getOutput()), encrypt(_encrypt)
{
init();
@@ -270,17 +282,17 @@ void CyrusAuthenticator::init()
NULL, /* Callbacks */
0, /* Connection flags */
&sasl_conn);
-
+
if (SASL_OK != code) {
QPID_LOG(error, "SASL: Connection creation failed: [" << code << "] " << sasl_errdetail(sasl_conn));
-
+
// TODO: Change this to an exception signaling
// server error, when one is available
throw ConnectionForcedException("Unable to perform authentication");
}
sasl_security_properties_t secprops;
-
+
//TODO: should the actual SSF values be configurable here?
secprops.min_ssf = encrypt ? 10: 0;
secprops.max_ssf = 256;
@@ -318,14 +330,14 @@ void CyrusAuthenticator::init()
secprops.property_values = 0;
secprops.security_flags = 0; /* or SASL_SEC_NOANONYMOUS etc as appropriate */
/*
- * The nodict flag restricts SASL authentication mechanisms
- * to those that are not susceptible to dictionary attacks.
- * They are:
+ * The nodict flag restricts SASL authentication mechanisms
+ * to those that are not susceptible to dictionary attacks.
+ * They are:
* SRP
* PASSDSS-3DES-1
* EXTERNAL
*/
- if (external.nodict) secprops.security_flags |= SASL_SEC_NODICTIONARY;
+ if (external.nodict) secprops.security_flags |= SASL_SEC_NODICTIONARY;
int result = sasl_setprop(sasl_conn, SASL_SEC_PROPS, &secprops);
if (result != SASL_OK) {
throw framing::InternalErrorException(QPID_MSG("SASL error: " << result));
@@ -370,10 +382,10 @@ void CyrusAuthenticator::getMechanisms(A
"", separator, "",
&list, &list_len,
&count);
-
+
if (SASL_OK != code) {
QPID_LOG(info, "SASL: Mechanism listing failed: " << sasl_errdetail(sasl_conn));
-
+
// TODO: Change this to an exception signaling
// server error, when one is available
throw ConnectionForcedException("Mechanism listing failed");
@@ -381,17 +393,17 @@ void CyrusAuthenticator::getMechanisms(A
string mechanism;
unsigned int start;
unsigned int end;
-
+
QPID_LOG(info, "SASL: Mechanism list: " << list);
-
+
end = 0;
do {
start = end;
-
+
// Seek to end of next mechanism
while (end < list_len && separator[0] != list[end])
end++;
-
+
// Record the mechanism
mechanisms.add(boost::shared_ptr<FieldValue>(new Str16Value(string(list, start, end - start))));
end++;
@@ -403,20 +415,20 @@ void CyrusAuthenticator::start(const str
{
const char *challenge;
unsigned int challenge_len;
-
+
// This should be at same debug level as mech list in getMechanisms().
QPID_LOG(info, "SASL: Starting authentication with mechanism: " << mechanism);
int code = sasl_server_start(sasl_conn,
mechanism.c_str(),
(response ? response->c_str() : 0), (response ? response->size() : 0),
&challenge, &challenge_len);
-
+
processAuthenticationStep(code, challenge, challenge_len);
qmf::org::apache::qpid::broker::Connection* cnxMgmt = connection.getMgmtObject();
- if ( cnxMgmt )
+ if ( cnxMgmt )
cnxMgmt->set_saslMechanism(mechanism);
}
-
+
void CyrusAuthenticator::step(const string& response)
{
const char *challenge;
@@ -438,10 +450,17 @@ void CyrusAuthenticator::processAuthenti
// authentication failure, when one is available
throw ConnectionForcedException("Authenticated username unavailable");
}
- QPID_LOG(info, connection.getMgmtId() << " SASL: Authentication succeeded for: " << uid);
connection.setUserId(uid);
+ AclModule* acl = connection.getBroker().getAcl();
+ if (acl && !acl->approveConnection(connection))
+ {
+ throw ConnectionForcedException("User connection denied by configured limit");
+ }
+
+ QPID_LOG(info, connection.getMgmtId() << " SASL: Authentication succeeded for: " << uid);
+
client.tune(framing::CHANNEL_MAX, connection.getFrameMax(), 0, connection.getHeartbeatMax());
} else if (SASL_CONTINUE == code) {
string challenge_str(challenge, challenge_len);
@@ -489,7 +508,7 @@ std::auto_ptr<SecurityLayer> CyrusAuthen
securityLayer = std::auto_ptr<SecurityLayer>(new CyrusSecurityLayer(sasl_conn, maxFrameSize));
}
qmf::org::apache::qpid::broker::Connection* cnxMgmt = connection.getMgmtObject();
- if ( cnxMgmt )
+ if ( cnxMgmt )
cnxMgmt->set_saslSsf(ssf);
return securityLayer;
}
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SaslAuthenticator.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SaslAuthenticator.h?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SaslAuthenticator.h (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SaslAuthenticator.h Thu Jun 28 09:14:52 2012
@@ -54,7 +54,7 @@ public:
static void init(const std::string& saslName, std::string const & saslConfigPath );
static void fini(void);
- static std::auto_ptr<SaslAuthenticator> createAuthenticator(Connection& connection, bool isShadow);
+ static std::auto_ptr<SaslAuthenticator> createAuthenticator(Connection& connection);
virtual void callUserIdCallbacks() { }
};
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SecureConnectionFactory.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SecureConnectionFactory.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SecureConnectionFactory.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SecureConnectionFactory.cpp Thu Jun 28 09:14:52 2012
@@ -7,9 +7,9 @@
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -41,11 +41,6 @@ SecureConnectionFactory::SecureConnectio
sys::ConnectionCodec*
SecureConnectionFactory::create(ProtocolVersion v, sys::OutputControl& out, const std::string& id,
const SecuritySettings& external) {
- if (broker.getConnectionCounter().allowConnection())
- {
- QPID_LOG(error, "Client max connection count limit exceeded: " << broker.getOptions().maxConnections << " connection refused");
- return 0;
- }
if (v == ProtocolVersion(0, 10)) {
SecureConnectionPtr sc(new SecureConnection());
CodecPtr c(new amqp_0_10::Connection(out, id, false));
@@ -71,5 +66,5 @@ SecureConnectionFactory::create(sys::Out
return sc.release();
}
-
+
}} // namespace qpid::broker
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SemanticState.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SemanticState.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SemanticState.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SemanticState.cpp Thu Jun 28 09:14:52 2012
@@ -72,7 +72,8 @@ SemanticState::SemanticState(DeliveryAda
dtxSelected(false),
authMsg(getSession().getBroker().getOptions().auth && !getSession().getConnection().isUserProxyAuth()),
userID(getSession().getConnection().getUserId()),
- closeComplete(false)
+ closeComplete(false),
+ connectionId(getSession().getConnection().getUrl())
{}
SemanticState::~SemanticState() {
@@ -142,6 +143,7 @@ bool SemanticState::cancel(const string&
DeliveryRecords::iterator removed =
remove_if(unacked.begin(), unacked.end(), bind(&DeliveryRecord::isRedundant, _1));
unacked.erase(removed, unacked.end());
+ getSession().setUnackedCount(unacked.size());
return true;
} else {
return false;
@@ -270,6 +272,7 @@ void SemanticState::checkDtxTimeout()
void SemanticState::record(const DeliveryRecord& delivery)
{
unacked.push_back(delivery);
+ getSession().setUnackedCount(unacked.size());
}
const std::string QPID_SYNC_FREQUENCY("qpid.sync_frequency");
@@ -426,7 +429,7 @@ void SemanticState::cancel(ConsumerImpl:
if(queue) {
queue->cancel(c);
if (queue->canAutoDelete() && !queue->hasExclusiveOwner()) {
- Queue::tryAutoDelete(session.getBroker(), queue);
+ Queue::tryAutoDelete(session.getBroker(), queue, connectionId, userID);
}
}
c->cancel();
@@ -555,6 +558,7 @@ void SemanticState::recover(bool requeue
//w.r.t id is lost
sort(unacked.begin(), unacked.end());
}
+ getSession().setUnackedCount(unacked.size());
}
void SemanticState::deliver(DeliveryRecord& msg, bool sync)
@@ -712,6 +716,7 @@ void SemanticState::release(DeliveryId f
DeliveryRecords::iterator removed =
remove_if(range.start, range.end, bind(&DeliveryRecord::isRedundant, _1));
unacked.erase(removed, range.end);
+ getSession().setUnackedCount(unacked.size());
}
void SemanticState::reject(DeliveryId first, DeliveryId last)
@@ -723,6 +728,7 @@ void SemanticState::reject(DeliveryId fi
if (i->isRedundant()) i = unacked.erase(i);
else i++;
}
+ getSession().setUnackedCount(unacked.size());
}
bool SemanticState::ConsumerImpl::doOutput()
@@ -810,6 +816,7 @@ void SemanticState::accepted(const Seque
(TransactionContext*) 0)));
unacked.erase(removed, unacked.end());
}
+ getSession().setUnackedCount(unacked.size());
}
void SemanticState::completed(const SequenceSet& commands) {
@@ -819,6 +826,7 @@ void SemanticState::completed(const Sequ
bind(&SemanticState::complete, this, _1)));
unacked.erase(removed, unacked.end());
requestDispatch();
+ getSession().setUnackedCount(unacked.size());
}
void SemanticState::attached()
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SemanticState.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SemanticState.h?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SemanticState.h (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/qpid/broker/SemanticState.h Thu Jun 28 09:14:52 2012
@@ -146,6 +146,8 @@ class SemanticState : private boost::non
std::string getResumeId() const { return resumeId; };
const std::string& getTag() const { return tag; }
uint64_t getResumeTtl() const { return resumeTtl; }
+ uint32_t getDeliveryCount() const { return deliveryCount; }
+ void setDeliveryCount(uint32_t _deliveryCount) { deliveryCount = _deliveryCount; }
const framing::FieldTable& getArguments() const { return arguments; }
SemanticState& getParent() { return *parent; }
@@ -180,6 +182,8 @@ class SemanticState : private boost::non
const bool authMsg;
const std::string userID;
bool closeComplete;
+ //needed for queue delete events in auto-delete:
+ const std::string connectionId;
void route(boost::intrusive_ptr<Message> msg, Deliverable& strategy);
void checkDtxTimeout();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org