You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2011/04/01 22:03:17 UTC
svn commit: r1087871 - in /qpid/branches/qpid-2920/qpid:
cpp/examples/old_api/ cpp/src/ cpp/src/qpid/broker/ cpp/src/qpid/sys/cyrus/
cpp/src/qpid/sys/posix/ cpp/src/tests/ doc/book/src/
java/broker/src/main/java/org/apache/qpid/server/ java/broker/src/...
Author: aconway
Date: Fri Apr 1 20:03:16 2011
New Revision: 1087871
URL: http://svn.apache.org/viewvc?rev=1087871&view=rev
Log:
Merge branch 'trunk' into qpid-2920
Added:
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRunner.java (with props)
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/CRAMMD5HexServerTest.java (with props)
Modified:
qpid/branches/qpid-2920/qpid/cpp/examples/old_api/Makefile.am
qpid/branches/qpid-2920/qpid/cpp/src/Makefile.am
qpid/branches/qpid-2920/qpid/cpp/src/qpid/broker/Queue.cpp
qpid/branches/qpid-2920/qpid/cpp/src/qpid/broker/QueuePolicy.cpp
qpid/branches/qpid-2920/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp
qpid/branches/qpid-2920/qpid/cpp/src/qpid/sys/posix/Socket.cpp
qpid/branches/qpid-2920/qpid/cpp/src/tests/brokertest.py
qpid/branches/qpid-2920/qpid/doc/book/src/Security.xml
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/message/ContentHeaderBodyAdapter.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServerFactory.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/anonymous/AnonymousSaslServerFactory.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HexInitialiser.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServerFactory.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java
qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AckTest.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java
qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java
qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java
qpid/branches/qpid-2920/qpid/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
qpid/branches/qpid-2920/qpid/tools/src/py/qpid-config
Modified: qpid/branches/qpid-2920/qpid/cpp/examples/old_api/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/cpp/examples/old_api/Makefile.am?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/cpp/examples/old_api/Makefile.am (original)
+++ qpid/branches/qpid-2920/qpid/cpp/examples/old_api/Makefile.am Fri Apr 1 20:03:16 2011
@@ -36,7 +36,7 @@ $(MAKEDIST): Makefile
examplesdir=$(pkgdatadir)/examples/old_api
dist_examples_DATA = $(MAKEDIST)
-EXTRA_DIST = README.verify verify verify_all
+EXTRA_DIST = README.verify verify verify_all CMakeLists.txt
# For older versions of automake
abs_top_srcdir = @abs_top_srcdir@
Modified: qpid/branches/qpid-2920/qpid/cpp/src/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/cpp/src/Makefile.am?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/cpp/src/Makefile.am (original)
+++ qpid/branches/qpid-2920/qpid/cpp/src/Makefile.am Fri Apr 1 20:03:16 2011
@@ -744,7 +744,7 @@ libqpidclient_la_SOURCES = \
QPIDCLIENT_VERSION_INFO = 2:0:0
libqpidclient_la_LDFLAGS = -version-info $(QPIDCLIENT_VERSION_INFO)
-libqpidtypes_la_libadd=-luuid
+libqpidtypes_la_LIBADD= -luuid
libqpidtypes_la_SOURCES= \
qpid/types/Exception.cpp \
qpid/types/Uuid.cpp \
Modified: qpid/branches/qpid-2920/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/cpp/src/qpid/broker/Queue.cpp?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/branches/qpid-2920/qpid/cpp/src/qpid/broker/Queue.cpp Fri Apr 1 20:03:16 2011
@@ -188,7 +188,6 @@ void Queue::recover(boost::intrusive_ptr
// setup synclist for recovered messages, so they don't get re-stored on lastNodeFailure
msg->addToSyncList(shared_from_this(), store);
}
- msg->enqueueComplete(); // mark the message as enqueued
if (store && (!msg->isContentLoaded() || msg->checkContentReleasable())) {
//content has not been loaded, need to ensure that lazy loading mode is set:
@@ -215,7 +214,6 @@ void Queue::requeue(const QueuedMessage&
{
Mutex::ScopedLock locker(messageLock);
if (!isEnqueued(msg)) return;
- msg.payload->enqueueComplete(); // mark the message as enqueued
messages->reinsert(msg);
listeners.populate(copy);
@@ -666,7 +664,9 @@ bool Queue::enqueue(TransactionContext*
}
if ((msg->isPersistent() || msg->checkContentReleasable()) && store) {
- msg->enqueueAsync(shared_from_this(), store); //increment to async counter -- for message sent to more than one queue
+ // mark the message as being enqueued - the store MUST CALL msg->enqueueComplete()
+ // when it considers the message stored.
+ msg->enqueueAsync(shared_from_this(), store);
boost::intrusive_ptr<PersistableMessage> pmsg = boost::static_pointer_cast<PersistableMessage>(msg);
store->enqueue(ctxt, pmsg, *this);
return true;
Modified: qpid/branches/qpid-2920/qpid/cpp/src/qpid/broker/QueuePolicy.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/cpp/src/qpid/broker/QueuePolicy.cpp?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/cpp/src/qpid/broker/QueuePolicy.cpp (original)
+++ qpid/branches/qpid-2920/qpid/cpp/src/qpid/broker/QueuePolicy.cpp Fri Apr 1 20:03:16 2011
@@ -246,7 +246,7 @@ bool RingQueuePolicy::checkLimit(boost::
{
// If the message is bigger than the queue size, give up
- if (m->contentSize() > getMaxSize()) {
+ if (getMaxSize() && m->contentSize() > getMaxSize()) {
QPID_LOG(debug, "Message too large for ring queue " << name
<< " [" << *this << "] "
<< ": message size = " << m->contentSize() << " bytes"
Modified: qpid/branches/qpid-2920/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp (original)
+++ qpid/branches/qpid-2920/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp Fri Apr 1 20:03:16 2011
@@ -106,7 +106,7 @@ size_t CyrusSecurityLayer::encode(const
bool CyrusSecurityLayer::canEncode()
{
- return encrypted || codec->canEncode();
+ return codec && (encrypted || codec->canEncode());
}
void CyrusSecurityLayer::init(qpid::sys::Codec* c)
Modified: qpid/branches/qpid-2920/qpid/cpp/src/qpid/sys/posix/Socket.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/cpp/src/qpid/sys/posix/Socket.cpp?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/cpp/src/qpid/sys/posix/Socket.cpp (original)
+++ qpid/branches/qpid-2920/qpid/cpp/src/qpid/sys/posix/Socket.cpp Fri Apr 1 20:03:16 2011
@@ -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
@@ -45,9 +45,9 @@ namespace sys {
namespace {
std::string getName(int fd, bool local, bool includeService = false)
{
- ::sockaddr_storage name; // big enough for any socket address
+ ::sockaddr_storage name; // big enough for any socket address
::socklen_t namelen = sizeof(name);
-
+
int result = -1;
if (local) {
result = ::getsockname(fd, (::sockaddr*)&name, &namelen);
@@ -60,8 +60,8 @@ std::string getName(int fd, bool local,
char servName[NI_MAXSERV];
char dispName[NI_MAXHOST];
if (includeService) {
- if (int rc=::getnameinfo((::sockaddr*)&name, namelen, dispName, sizeof(dispName),
- servName, sizeof(servName),
+ if (int rc=::getnameinfo((::sockaddr*)&name, namelen, dispName, sizeof(dispName),
+ servName, sizeof(servName),
NI_NUMERICHOST | NI_NUMERICSERV) != 0)
throw QPID_POSIX_ERROR(rc);
return std::string(dispName) + ":" + std::string(servName);
@@ -75,9 +75,9 @@ std::string getName(int fd, bool local,
std::string getService(int fd, bool local)
{
- ::sockaddr_storage name; // big enough for any socket address
+ ::sockaddr_storage name; // big enough for any socket address
::socklen_t namelen = sizeof(name);
-
+
int result = -1;
if (local) {
result = ::getsockname(fd, (::sockaddr*)&name, &namelen);
@@ -88,8 +88,8 @@ std::string getService(int fd, bool loca
QPID_POSIX_CHECK(result);
char servName[NI_MAXSERV];
- if (int rc=::getnameinfo((::sockaddr*)&name, namelen, 0, 0,
- servName, sizeof(servName),
+ if (int rc=::getnameinfo((::sockaddr*)&name, namelen, 0, 0,
+ servName, sizeof(servName),
NI_NUMERICHOST | NI_NUMERICSERV) != 0)
throw QPID_POSIX_ERROR(rc);
return servName;
@@ -172,6 +172,23 @@ void Socket::connect(const SocketAddress
(errno != EINPROGRESS)) {
throw Exception(QPID_MSG(strError(errno) << ": " << connectname));
}
+ // When connecting to a port on the same host which no longer has
+ // a process associated with it, the OS occasionally chooses the
+ // remote port (which is unoccupied) as the port to bind the local
+ // end of the socket, resulting in a "circular" connection.
+ //
+ // This seems like something the OS should prevent but I have
+ // confirmed that sporadic hangs in
+ // cluster_tests.LongTests.test_failover on RHEL5 are caused by
+ // such a circular connection.
+ //
+ // Raise an error if we see such a connection, since we know there is
+ // no listener on the peer address.
+ //
+ if (getLocalAddress() == getPeerAddress()) {
+ close();
+ throw Exception(QPID_MSG("Connection refused: " << connectname));
+ }
}
void
Modified: qpid/branches/qpid-2920/qpid/cpp/src/tests/brokertest.py
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/cpp/src/tests/brokertest.py?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/cpp/src/tests/brokertest.py (original)
+++ qpid/branches/qpid-2920/qpid/cpp/src/tests/brokertest.py Fri Apr 1 20:03:16 2011
@@ -509,7 +509,7 @@ class BrokerTest(TestCase):
r.close()
self.assertEqual(expect_contents, actual_contents)
-def join(thread, timeout=1):
+def join(thread, timeout=10):
thread.join(timeout)
if thread.isAlive(): raise Exception("Timed out joining thread %s"%thread)
Modified: qpid/branches/qpid-2920/qpid/doc/book/src/Security.xml
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/doc/book/src/Security.xml?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/doc/book/src/Security.xml (original)
+++ qpid/branches/qpid-2920/qpid/doc/book/src/Security.xml Fri Apr 1 20:03:16 2011
@@ -1,5 +1,25 @@
<?xml version='1.0' encoding='utf-8' ?>
<!DOCTYPE chapter PUBLIC "-//OASIS//DTD DocBook XML V4.5//EN" "http://www.oasis-open.org/docbook/xml/4.5/docbookx.dtd">
+<!--
+
+ 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
+ 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
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
<section id="chap-Messaging_User_Guide-Security">
<title>Security</title>
<para>
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java Fri Apr 1 20:03:16 2011
@@ -346,7 +346,7 @@ public class AMQChannel implements Sessi
finally
{
long bodySize = _currentMessage.getSize();
- long timestamp = ((BasicContentHeaderProperties) _currentMessage.getContentHeader().properties).getTimestamp();
+ long timestamp = ((BasicContentHeaderProperties) _currentMessage.getContentHeader().getProperties()).getTimestamp();
_session.registerMessageReceived(bodySize, timestamp);
_currentMessage = null;
}
@@ -1079,8 +1079,8 @@ public class AMQChannel implements Sessi
private boolean checkMessageUserId(ContentHeaderBody header)
{
AMQShortString userID =
- header.properties instanceof BasicContentHeaderProperties
- ? ((BasicContentHeaderProperties) header.properties).getUserId()
+ header.getProperties() instanceof BasicContentHeaderProperties
+ ? ((BasicContentHeaderProperties) header.getProperties()).getUserId()
: null;
return (!MSG_AUTH || _session.getPrincipal().getName().equals(userID == null? "" : userID.toString()));
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/message/ContentHeaderBodyAdapter.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/message/ContentHeaderBodyAdapter.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/message/ContentHeaderBodyAdapter.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/message/ContentHeaderBodyAdapter.java Fri Apr 1 20:03:16 2011
@@ -37,7 +37,7 @@ public class ContentHeaderBodyAdapter im
private BasicContentHeaderProperties getProperties()
{
- return (BasicContentHeaderProperties) _contentHeaderBody.properties;
+ return (BasicContentHeaderProperties) _contentHeaderBody.getProperties();
}
public String getCorrelationId()
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData.java Fri Apr 1 20:03:16 2011
@@ -161,7 +161,7 @@ public class MessageMetaData implements
public boolean isPersistent()
{
- BasicContentHeaderProperties properties = (BasicContentHeaderProperties) (_contentHeaderBody.properties);
+ BasicContentHeaderProperties properties = (BasicContentHeaderProperties) (_contentHeaderBody.getProperties());
return properties.getDeliveryMode() == BasicContentHeaderProperties.PERSISTENT;
}
@@ -229,7 +229,7 @@ public class MessageMetaData implements
{
private BasicContentHeaderProperties getProperties()
{
- return (BasicContentHeaderProperties) getContentHeaderBody().properties;
+ return (BasicContentHeaderProperties) getContentHeaderBody().getProperties();
}
public String getCorrelationId()
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java Fri Apr 1 20:03:16 2011
@@ -507,7 +507,7 @@ public class AMQQueueMBean extends AMQMa
private String[] getMessageHeaderProperties(ContentHeaderBody headerBody)
{
List<String> list = new ArrayList<String>();
- BasicContentHeaderProperties headerProperties = (BasicContentHeaderProperties) headerBody.properties;
+ BasicContentHeaderProperties headerProperties = (BasicContentHeaderProperties) headerBody.getProperties();
list.add("reply-to = " + headerProperties.getReplyToAsString());
list.add("propertyFlags = " + headerProperties.getPropertyFlags());
list.add("ApplicationID = " + headerProperties.getAppIdAsString());
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java Fri Apr 1 20:03:16 2011
@@ -96,9 +96,9 @@ public class IncomingMessage implements
public void setExpiration()
{
long expiration =
- ((BasicContentHeaderProperties) _contentHeaderBody.properties).getExpiration();
+ ((BasicContentHeaderProperties) _contentHeaderBody.getProperties()).getExpiration();
long timestamp =
- ((BasicContentHeaderProperties) _contentHeaderBody.properties).getTimestamp();
+ ((BasicContentHeaderProperties) _contentHeaderBody.getProperties()).getTimestamp();
if (SYNCHED_CLOCKS)
{
@@ -193,8 +193,8 @@ public class IncomingMessage implements
public boolean isPersistent()
{
- return getContentHeader().properties instanceof BasicContentHeaderProperties &&
- ((BasicContentHeaderProperties) getContentHeader().properties).getDeliveryMode() ==
+ return getContentHeader().getProperties() instanceof BasicContentHeaderProperties &&
+ ((BasicContentHeaderProperties) getContentHeader().getProperties()).getDeliveryMode() ==
BasicContentHeaderProperties.PERSISTENT;
}
Added: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRunner.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRunner.java?rev=1087871&view=auto
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRunner.java (added)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRunner.java Fri Apr 1 20:03:16 2011
@@ -0,0 +1,79 @@
+/*
+ *
+ * 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
+ * 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
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.queue;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.pool.ReadWriteRunnable;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.queue.QueueRunner;
+import org.apache.qpid.server.queue.SimpleAMQQueue;
+
+
+public class QueueRunner implements ReadWriteRunnable
+{
+ private static final Logger _logger = Logger.getLogger(QueueRunner.class);
+
+ private String _name;
+ private SimpleAMQQueue _queue;
+
+ public QueueRunner(SimpleAMQQueue queue, long count)
+ {
+ _queue = queue;
+ _name = "QueueRunner-" + count + "-" + queue.getLogActor();
+ }
+
+ public void run()
+ {
+ String originalName = Thread.currentThread().getName();
+ try
+ {
+ Thread.currentThread().setName(_name);
+ CurrentActor.set(_queue.getLogActor());
+
+ _queue.processQueue(this);
+ }
+ catch (AMQException e)
+ {
+ _logger.error(e);
+ }
+ finally
+ {
+ CurrentActor.remove();
+ Thread.currentThread().setName(originalName);
+ }
+ }
+
+ public boolean isRead()
+ {
+ return false;
+ }
+
+ public boolean isWrite()
+ {
+ return true;
+ }
+
+ public String toString()
+ {
+ return _name;
+ }
+}
\ No newline at end of file
Propchange: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRunner.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRunner.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java Fri Apr 1 20:03:16 2011
@@ -1585,7 +1585,7 @@ public class SimpleAMQQueue implements A
public void deliverAsync()
{
- Runner runner = new Runner(_stateChangeCount.incrementAndGet());
+ QueueRunner runner = new QueueRunner(this, _stateChangeCount.incrementAndGet());
if (_asynchronousRunner.compareAndSet(null, runner))
{
@@ -1604,52 +1604,6 @@ public class SimpleAMQQueue implements A
_asyncDelivery.execute(flusher);
}
-
- private class Runner implements ReadWriteRunnable
- {
- String _name;
- public Runner(long count)
- {
- _name = "QueueRunner-" + count + "-" + _logActor;
- }
-
- public void run()
- {
- String originalName = Thread.currentThread().getName();
- try
- {
- Thread.currentThread().setName(_name);
- CurrentActor.set(_logActor);
-
- processQueue(this);
- }
- catch (AMQException e)
- {
- _logger.error(e);
- }
- finally
- {
- CurrentActor.remove();
- Thread.currentThread().setName(originalName);
- }
- }
-
- public boolean isRead()
- {
- return false;
- }
-
- public boolean isWrite()
- {
- return true;
- }
-
- public String toString()
- {
- return _name;
- }
- }
-
public void flushSubscription(Subscription sub) throws AMQException
{
// Access control
@@ -1834,7 +1788,7 @@ public class SimpleAMQQueue implements A
* @param runner the Runner to schedule
* @throws AMQException
*/
- private void processQueue(Runnable runner) throws AMQException
+ public void processQueue(QueueRunner runner) throws AMQException
{
long stateChangeCount;
long previousStateChangeCount = Long.MIN_VALUE;
@@ -2289,4 +2243,9 @@ public class SimpleAMQQueue implements A
}
}
}
+
+ public LogActor getLogActor()
+ {
+ return _logActor;
+ }
}
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServerFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServerFactory.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServerFactory.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServerFactory.java Fri Apr 1 20:03:16 2011
@@ -45,9 +45,10 @@ public class AmqPlainSaslServerFactory i
public String[] getMechanismNames(Map props)
{
- if (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
- props.containsKey(Sasl.POLICY_NODICTIONARY) ||
- props.containsKey(Sasl.POLICY_NOACTIVE))
+ if (props != null &&
+ (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
+ props.containsKey(Sasl.POLICY_NODICTIONARY) ||
+ props.containsKey(Sasl.POLICY_NOACTIVE)))
{
// returned array must be non null according to interface documentation
return new String[0];
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/anonymous/AnonymousSaslServerFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/anonymous/AnonymousSaslServerFactory.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/anonymous/AnonymousSaslServerFactory.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/anonymous/AnonymousSaslServerFactory.java Fri Apr 1 20:03:16 2011
@@ -47,10 +47,11 @@ public class AnonymousSaslServerFactory
public String[] getMechanismNames(Map props)
{
- if (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
- props.containsKey(Sasl.POLICY_NODICTIONARY) ||
- props.containsKey(Sasl.POLICY_NOACTIVE) ||
- props.containsKey(Sasl.POLICY_NOANONYMOUS))
+ if (props != null &&
+ (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
+ props.containsKey(Sasl.POLICY_NODICTIONARY) ||
+ props.containsKey(Sasl.POLICY_NOACTIVE) ||
+ props.containsKey(Sasl.POLICY_NOANONYMOUS)))
{
// returned array must be non null according to interface documentation
return new String[0];
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HexInitialiser.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HexInitialiser.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HexInitialiser.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HexInitialiser.java Fri Apr 1 20:03:16 2011
@@ -70,7 +70,7 @@ public class CRAMMD5HexInitialiser exten
for (char c : password)
{
//toHexString does not prepend 0 so we have to
- if (((byte) c > -1) && (byte) c < 10)
+ if (((byte) c > -1) && (byte) c < 0x10 )
{
sb.append(0);
}
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServerFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServerFactory.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServerFactory.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServerFactory.java Fri Apr 1 20:03:16 2011
@@ -45,9 +45,10 @@ public class PlainSaslServerFactory impl
public String[] getMechanismNames(Map props)
{
- if (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
- props.containsKey(Sasl.POLICY_NODICTIONARY) ||
- props.containsKey(Sasl.POLICY_NOACTIVE))
+ if (props != null &&
+ (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
+ props.containsKey(Sasl.POLICY_NODICTIONARY) ||
+ props.containsKey(Sasl.POLICY_NOACTIVE)))
{
// returned array must be non null according to interface documentation
return new String[0];
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java Fri Apr 1 20:03:16 2011
@@ -441,7 +441,7 @@ public class Subscription_0_10 implement
Struct[] headers = new Struct[] { deliveryProps, messageProps };
BasicContentHeaderProperties properties =
- (BasicContentHeaderProperties) message_0_8.getContentHeaderBody().properties;
+ (BasicContentHeaderProperties) message_0_8.getContentHeaderBody().getProperties();
final AMQShortString exchange = message_0_8.getMessagePublishInfo().getExchange();
if(exchange != null)
{
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java Fri Apr 1 20:03:16 2011
@@ -364,7 +364,7 @@ public class Show extends AbstractComman
{
if(msg instanceof AMQMessage)
{
- headers = ((BasicContentHeaderProperties) ((AMQMessage)msg).getContentHeaderBody().properties);
+ headers = ((BasicContentHeaderProperties) ((AMQMessage)msg).getContentHeaderBody().getProperties());
}
}
catch (AMQException e)
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java Fri Apr 1 20:03:16 2011
@@ -276,7 +276,7 @@ public class AbstractHeadersExchangeTest
static ContentHeaderBody getContentHeader(FieldTable headers)
{
ContentHeaderBody header = new ContentHeaderBody();
- header.properties = getProperties(headers);
+ header.setProperties(getProperties(headers));
return header;
}
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java Fri Apr 1 20:03:16 2011
@@ -396,7 +396,7 @@ public class TopicExchangeTest extends I
IncomingMessage message = new IncomingMessage(info);
final ContentHeaderBody chb = new ContentHeaderBody();
BasicContentHeaderProperties props = new BasicContentHeaderProperties();
- chb.properties = props;
+ chb.setProperties(props);
message.setContentHeaderBody(chb);
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java Fri Apr 1 20:03:16 2011
@@ -96,7 +96,7 @@ public class AMQPriorityQueueTest extend
AMQMessage msg = super.createMessage(id);
BasicContentHeaderProperties props = new BasicContentHeaderProperties();
props.setPriority(i);
- msg.getContentHeaderBody().properties = props;
+ msg.getContentHeaderBody().setProperties(props);
return msg;
}
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java Fri Apr 1 20:03:16 2011
@@ -277,7 +277,7 @@ public class AMQQueueAlertTest extends I
ContentHeaderBody contentHeaderBody = new ContentHeaderBody();
BasicContentHeaderProperties props = new BasicContentHeaderProperties();
- contentHeaderBody.properties = props;
+ contentHeaderBody.setProperties(props);
contentHeaderBody.bodySize = size; // in bytes
IncomingMessage message = new IncomingMessage(publish);
message.setContentHeaderBody(contentHeaderBody);
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java Fri Apr 1 20:03:16 2011
@@ -402,8 +402,8 @@ public class AMQQueueMBeanTest extends I
ContentHeaderBody contentHeaderBody = new ContentHeaderBody();
contentHeaderBody.bodySize = MESSAGE_SIZE; // in bytes
- contentHeaderBody.properties = new BasicContentHeaderProperties();
- ((BasicContentHeaderProperties) contentHeaderBody.properties).setDeliveryMode((byte) (persistent ? 2 : 1));
+ contentHeaderBody.setProperties(new BasicContentHeaderProperties());
+ ((BasicContentHeaderProperties) contentHeaderBody.getProperties()).setDeliveryMode((byte) (persistent ? 2 : 1));
IncomingMessage msg = new IncomingMessage(publish);
msg.setContentHeaderBody(contentHeaderBody);
return msg;
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AckTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AckTest.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AckTest.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AckTest.java Fri Apr 1 20:03:16 2011
@@ -126,7 +126,7 @@ public class AckTest extends InternalBro
//IncomingMessage msg2 = null;
BasicContentHeaderProperties b = new BasicContentHeaderProperties();
ContentHeaderBody cb = new ContentHeaderBody();
- cb.properties = b;
+ cb.setProperties(b);
if (persistent)
{
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java Fri Apr 1 20:03:16 2011
@@ -660,8 +660,8 @@ public class SimpleAMQQueueTest extends
// Create IncomingMessage and nondurable queue
final IncomingMessage msg = new IncomingMessage(info);
ContentHeaderBody contentHeaderBody = new ContentHeaderBody();
- contentHeaderBody.properties = new BasicContentHeaderProperties();
- ((BasicContentHeaderProperties) contentHeaderBody.properties).setDeliveryMode((byte) 2);
+ contentHeaderBody.setProperties(new BasicContentHeaderProperties());
+ ((BasicContentHeaderProperties) contentHeaderBody.getProperties()).setDeliveryMode((byte) 2);
msg.setContentHeaderBody(contentHeaderBody);
final ArrayList<BaseQueue> qs = new ArrayList<BaseQueue>();
@@ -707,6 +707,111 @@ public class SimpleAMQQueueTest extends
}
+ /**
+ * processQueue() is used when asynchronously delivering messages to
+ * subscriptions which could not be delivered immediately during the
+ * enqueue() operation.
+ *
+ * A defect within the method would mean that delivery of these messages may
+ * not occur should the Runner stop before all messages have been processed.
+ * Such a defect was discovered when Selectors were used such that one and
+ * only one subscription can/will accept any given messages, but multiple
+ * subscriptions are present, and one of the earlier subscriptions receives
+ * more messages than the others.
+ *
+ * This test is to validate that the processQueue() method is able to
+ * correctly deliver all of the messages present for asynchronous delivery
+ * to subscriptions in such a scenario.
+ */
+ public void testProcessQueueWithUniqueSelectors() throws Exception
+ {
+ TestSimpleQueueEntryListFactory factory = new TestSimpleQueueEntryListFactory();
+ SimpleAMQQueue testQueue = new SimpleAMQQueue("testQueue", false, "testOwner",false,
+ false, _virtualHost, factory, null)
+ {
+ @Override
+ public void deliverAsync(Subscription sub)
+ {
+ // do nothing, i.e prevent deliveries by the SubFlushRunner
+ // when registering the new subscriptions
+ }
+ };
+
+ // retrieve the QueueEntryList the queue creates and insert the test
+ // messages, thus avoiding straight-through delivery attempts during
+ //enqueue() process.
+ QueueEntryList list = factory.getQueueEntryList();
+ assertNotNull("QueueEntryList should have been created", list);
+
+ QueueEntry msg1 = list.add(createMessage(1L));
+ QueueEntry msg2 = list.add(createMessage(2L));
+ QueueEntry msg3 = list.add(createMessage(3L));
+ QueueEntry msg4 = list.add(createMessage(4L));
+ QueueEntry msg5 = list.add(createMessage(5L));
+
+ // Create lists of the entries each subscription should be interested
+ // in.Bias over 50% of the messages to the first subscription so that
+ // the later subscriptions reject them and report being done before
+ // the first subscription as the processQueue method proceeds.
+ List<QueueEntry> msgListSub1 = createEntriesList(msg1, msg2, msg3);
+ List<QueueEntry> msgListSub2 = createEntriesList(msg4);
+ List<QueueEntry> msgListSub3 = createEntriesList(msg5);
+
+ MockSubscription sub1 = new MockSubscription(msgListSub1);
+ MockSubscription sub2 = new MockSubscription(msgListSub2);
+ MockSubscription sub3 = new MockSubscription(msgListSub3);
+
+ // register the subscriptions
+ testQueue.registerSubscription(sub1, false);
+ testQueue.registerSubscription(sub2, false);
+ testQueue.registerSubscription(sub3, false);
+
+ //check that no messages have been delivered to the
+ //subscriptions during registration
+ assertEquals("No messages should have been delivered yet", 0, sub1.getMessages().size());
+ assertEquals("No messages should have been delivered yet", 0, sub2.getMessages().size());
+ assertEquals("No messages should have been delivered yet", 0, sub3.getMessages().size());
+
+ // call processQueue to deliver the messages
+ testQueue.processQueue(new QueueRunner(testQueue, 1)
+ {
+ @Override
+ public void run()
+ {
+ // we dont actually want/need this runner to do any work
+ // because we we are already doing it!
+ }
+ });
+
+ // check expected messages delivered to correct consumers
+ verifyRecievedMessages(msgListSub1, sub1.getMessages());
+ verifyRecievedMessages(msgListSub2, sub2.getMessages());
+ verifyRecievedMessages(msgListSub3, sub3.getMessages());
+ }
+
+ private List<QueueEntry> createEntriesList(QueueEntry... entries)
+ {
+ ArrayList<QueueEntry> entriesList = new ArrayList<QueueEntry>();
+ for (QueueEntry entry : entries)
+ {
+ entriesList.add(entry);
+ }
+ return entriesList;
+ }
+
+ private void verifyRecievedMessages(List<QueueEntry> expected,
+ List<QueueEntry> delivered)
+ {
+ assertEquals("Consumer did not receive the expected number of messages",
+ expected.size(), delivered.size());
+
+ for (QueueEntry msg : expected)
+ {
+ assertTrue("Consumer did not recieve msg: "
+ + msg.getMessage().getMessageNumber(), delivered.contains(msg));
+ }
+ }
+
public class TestMessage extends AMQMessage
{
private final long _tag;
@@ -747,4 +852,20 @@ public class SimpleAMQQueueTest extends
AMQMessage messageA = new TestMessage(id, id, info);
return messageA;
}
+
+ class TestSimpleQueueEntryListFactory implements QueueEntryListFactory
+ {
+ QueueEntryList _list;
+
+ public QueueEntryList createQueueEntryList(AMQQueue queue)
+ {
+ _list = new SimpleQueueEntryList(queue);
+ return _list;
+ }
+
+ public QueueEntryList getQueueEntryList()
+ {
+ return _list;
+ }
+ }
}
Added: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/CRAMMD5HexServerTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/CRAMMD5HexServerTest.java?rev=1087871&view=auto
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/CRAMMD5HexServerTest.java (added)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/CRAMMD5HexServerTest.java Fri Apr 1 20:03:16 2011
@@ -0,0 +1,230 @@
+/*
+ *
+ * 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
+ * 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
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.server.security.auth.sasl;
+
+import java.io.File;
+import java.io.IOException;
+import java.security.MessageDigest;
+import java.security.Principal;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import javax.security.auth.login.AccountNotFoundException;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+
+import junit.framework.TestCase;
+
+import org.apache.commons.codec.binary.Hex;
+import org.apache.qpid.server.security.auth.database.Base64MD5PasswordFilePrincipalDatabase;
+import org.apache.qpid.server.security.auth.sasl.crammd5.CRAMMD5HexInitialiser;
+import org.apache.qpid.server.security.auth.sasl.crammd5.CRAMMD5HexSaslServer;
+import org.apache.qpid.server.security.auth.sasl.crammd5.CRAMMD5HexServerFactory;
+
+/**
+ * Test for the CRAM-MD5-HEX SASL mechanism.
+ *
+ * This test case focuses on testing {@link CRAMMD5HexSaslServer} but also exercises
+ * collaborators {@link CRAMMD5HexInitialiser} and {@link Base64MD5PasswordFilePrincipalDatabase}
+ */
+public class CRAMMD5HexServerTest extends TestCase
+{
+
+ private SaslServer _saslServer; // Class under test
+ private CRAMMD5HexServerFactory _saslFactory;
+
+ @Override
+ protected void setUp() throws Exception
+ {
+ super.setUp();
+
+ CRAMMD5HexInitialiser _initializer = new CRAMMD5HexInitialiser();
+
+ //Use properties to create a PrincipalDatabase
+ Base64MD5PasswordFilePrincipalDatabase db = createTestPrincipalDatabase();
+ assertEquals("Unexpected number of test users in the db", 2, db.getUsers().size());
+
+ _initializer.initialise(db);
+
+ _saslFactory = new CRAMMD5HexServerFactory();
+
+ _saslServer = _saslFactory.createSaslServer(CRAMMD5HexSaslServer.MECHANISM,
+ "AMQP",
+ "localhost",
+ _initializer.getProperties(),
+ _initializer.getCallbackHandler());
+ assertNotNull("Unable to create saslServer with mechanism type " + CRAMMD5HexSaslServer.MECHANISM, _saslServer);
+
+ }
+
+ public void testSuccessfulAuth() throws Exception
+ {
+
+ final byte[] serverChallenge = _saslServer.evaluateResponse(new byte[0]);
+
+ // Generate client response
+ final byte[] clientResponse = generateClientResponse("knownuser", "guest", serverChallenge);
+
+
+ byte[] nextServerChallenge = _saslServer.evaluateResponse(clientResponse);
+ assertTrue("Exchange must be flagged as complete after successful authentication", _saslServer.isComplete());
+ assertNull("Next server challenge must be null after successful authentication", nextServerChallenge);
+
+ }
+
+ public void testKnownUserPresentsWrongPassword() throws Exception
+ {
+ byte[] serverChallenge = _saslServer.evaluateResponse(new byte[0]);
+
+
+ final byte[] clientResponse = generateClientResponse("knownuser", "wrong!", serverChallenge);
+ try
+ {
+ _saslServer.evaluateResponse(clientResponse);
+ fail("Exception not thrown");
+ }
+ catch (SaslException se)
+ {
+ // PASS
+ }
+ assertFalse("Exchange must not be flagged as complete after unsuccessful authentication", _saslServer.isComplete());
+ }
+
+ public void testUnknownUser() throws Exception
+ {
+ final byte[] serverChallenge = _saslServer.evaluateResponse(new byte[0]);
+
+
+ final byte[] clientResponse = generateClientResponse("unknownuser", "guest", serverChallenge);
+
+ try
+ {
+ _saslServer.evaluateResponse(clientResponse);
+ fail("Exception not thrown");
+ }
+ catch (SaslException se)
+ {
+ assertExceptionHasUnderlyingAsCause(AccountNotFoundException.class, se);
+ // PASS
+ }
+ assertFalse("Exchange must not be flagged as complete after unsuccessful authentication", _saslServer.isComplete());
+ }
+
+ /**
+ *
+ * Demonstrates QPID-3158. A defect meant that users with some valid password were failing to
+ * authenticate when using the .NET 0-8 client (uses this SASL mechanism).
+ * It so happens that password "guest2" was one of the affected passwords.
+ *
+ * @throws Exception
+ */
+ public void testSuccessfulAuthReproducingQpid3158() throws Exception
+ {
+ byte[] serverChallenge = _saslServer.evaluateResponse(new byte[0]);
+
+ // Generate client response
+ byte[] resp = generateClientResponse("qpid3158user", "guest2", serverChallenge);
+
+ byte[] nextServerChallenge = _saslServer.evaluateResponse(resp);
+ assertTrue("Exchange must be flagged as complete after successful authentication", _saslServer.isComplete());
+ assertNull("Next server challenge must be null after successful authentication", nextServerChallenge);
+ }
+
+ /**
+ * Since we don't have a CRAM-MD5-HEX implementation client implementation in Java, this method
+ * provides the implementation for first principals.
+ *
+ * @param userId user id
+ * @param clearTextPassword clear text password
+ * @param serverChallenge challenge from server
+ *
+ * @return challenge response
+ */
+ private byte[] generateClientResponse(final String userId, final String clearTextPassword, final byte[] serverChallenge) throws Exception
+ {
+ byte[] digestedPasswordBytes = MessageDigest.getInstance("MD5").digest(clearTextPassword.getBytes());
+ char[] hexEncodedDigestedPassword = Hex.encodeHex(digestedPasswordBytes);
+ byte[] hexEncodedDigestedPasswordBytes = new String(hexEncodedDigestedPassword).getBytes();
+
+
+ Mac hmacMd5 = Mac.getInstance("HmacMD5");
+ hmacMd5.init(new SecretKeySpec(hexEncodedDigestedPasswordBytes, "HmacMD5"));
+ final byte[] messageAuthenticationCode = hmacMd5.doFinal(serverChallenge);
+
+ // Build client response
+ String responseAsString = userId + " " + new String(Hex.encodeHex(messageAuthenticationCode));
+ byte[] resp = responseAsString.getBytes();
+ return resp;
+ }
+
+ /**
+ * Creates a test principal database.
+ *
+ * @return
+ * @throws IOException
+ */
+ private Base64MD5PasswordFilePrincipalDatabase createTestPrincipalDatabase() throws IOException
+ {
+ Base64MD5PasswordFilePrincipalDatabase db = new Base64MD5PasswordFilePrincipalDatabase();
+ File file = File.createTempFile("passwd", "db");
+ file.deleteOnExit();
+ db.setPasswordFile(file.getCanonicalPath());
+ db.createPrincipal( createTestPrincipal("knownuser"), "guest".toCharArray());
+ db.createPrincipal( createTestPrincipal("qpid3158user"), "guest2".toCharArray());
+ return db;
+ }
+
+ private Principal createTestPrincipal(final String name)
+ {
+ return new Principal()
+ {
+
+ @Override
+ public String getName()
+ {
+ return name;
+ }
+ };
+ }
+
+ private void assertExceptionHasUnderlyingAsCause(final Class<? extends Throwable> expectedUnderlying, Throwable e)
+ {
+ assertNotNull(e);
+ int infiniteLoopGuard = 0; // Guard against loops in the cause chain
+ boolean foundExpectedUnderlying = false;
+ while (e.getCause() != null && infiniteLoopGuard++ < 10)
+ {
+ if (expectedUnderlying.equals(e.getCause().getClass()))
+ {
+ foundExpectedUnderlying = true;
+ break;
+ }
+ e = e.getCause();
+ }
+
+ if (!foundExpectedUnderlying)
+ {
+ fail("Not found expected underlying exception " + expectedUnderlying + " as underlying cause of " + e.getClass());
+ }
+ }
+
+}
Propchange: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/CRAMMD5HexServerTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/CRAMMD5HexServerTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java Fri Apr 1 20:03:16 2011
@@ -589,7 +589,7 @@ public class MessageStoreTest extends In
headerBody.classId = BasicConsumeBodyImpl.CLASS_ID;
headerBody.bodySize = 0;
- headerBody.properties = properties;
+ headerBody.setProperties(properties);
try
{
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java Fri Apr 1 20:03:16 2011
@@ -102,7 +102,7 @@ public class ReferenceCountingTest exten
ContentHeaderBody chb = new ContentHeaderBody();
BasicContentHeaderProperties bchp = new BasicContentHeaderProperties();
bchp.setDeliveryMode((byte)2);
- chb.properties = bchp;
+ chb.setProperties(bchp);
return chb;
}
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java Fri Apr 1 20:03:16 2011
@@ -22,6 +22,7 @@ package org.apache.qpid.server.subscript
*/
import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -45,6 +46,7 @@ public class MockSubscription implements
private State _state = State.ACTIVE;
private ArrayList<QueueEntry> messages = new ArrayList<QueueEntry>();
private final Lock _stateChangeLock = new ReentrantLock();
+ private List<QueueEntry> _acceptEntries = null;
private final QueueEntry.SubscriptionAcquiredState _owningState = new QueueEntry.SubscriptionAcquiredState(this);
private final QueueEntry.SubscriptionAssignedState _assignedState = new QueueEntry.SubscriptionAssignedState(this);
@@ -54,6 +56,15 @@ public class MockSubscription implements
// Create a simple ID that increments for ever new Subscription
private final long _subscriptionID = idGenerator.getAndIncrement();
+ public MockSubscription()
+ {
+ }
+
+ public MockSubscription(List<QueueEntry> acceptEntries)
+ {
+ _acceptEntries = acceptEntries;
+ }
+
public void close()
{
_closed = true;
@@ -119,8 +130,15 @@ public class MockSubscription implements
_stateChangeLock.lock();
}
- public boolean hasInterest(QueueEntry msg)
+ public boolean hasInterest(QueueEntry entry)
{
+ if(_acceptEntries != null)
+ {
+ //simulate selector behaviour, only signal
+ //interest in the dictated queue entries
+ return _acceptEntries.contains(entry);
+ }
+
return true;
}
Modified: qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java (original)
+++ qpid/branches/qpid-2920/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java Fri Apr 1 20:03:16 2011
@@ -243,7 +243,7 @@ public class InternalBrokerBaseCase exte
//Make Message Persistent
properties.setDeliveryMode((byte) 2);
- _headerBody.properties = properties;
+ _headerBody.setProperties(properties);
channel.publishContentHeader(_headerBody);
}
Modified: qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java (original)
+++ qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java Fri Apr 1 20:03:16 2011
@@ -1,3 +1,23 @@
+/*
+ *
+ * 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
+ * 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
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
package org.apache.qpid.client;
import java.util.ArrayList;
Modified: qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java (original)
+++ qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java Fri Apr 1 20:03:16 2011
@@ -99,7 +99,7 @@ public abstract class AbstractJMSMessage
}
AMQMessageDelegate delegate = new AMQMessageDelegate_0_8(messageNbr,
- (BasicContentHeaderProperties) contentHeader.properties,
+ (BasicContentHeaderProperties) contentHeader.getProperties(),
exchange, routingKey);
return createMessage(delegate, data);
Modified: qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java (original)
+++ qpid/branches/qpid-2920/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java Fri Apr 1 20:03:16 2011
@@ -104,7 +104,7 @@ public class MessageFactoryRegistry
AMQShortString routingKey, ContentHeaderBody contentHeader, List bodies)
throws AMQException, JMSException
{
- BasicContentHeaderProperties properties = (BasicContentHeaderProperties) contentHeader.properties;
+ BasicContentHeaderProperties properties = (BasicContentHeaderProperties) contentHeader.getProperties();
// Get the message content type. This may be null for pure AMQP messages, but will always be set for JMS over
// AMQP. When the type is null, it can only be assumed that the message is a byte message.
Modified: qpid/branches/qpid-2920/qpid/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java (original)
+++ qpid/branches/qpid-2920/qpid/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java Fri Apr 1 20:03:16 2011
@@ -36,7 +36,7 @@ public class ContentHeaderBody implement
public long bodySize;
/** must never be null */
- public ContentHeaderProperties properties;
+ private ContentHeaderProperties properties;
public ContentHeaderBody()
{
@@ -128,4 +128,14 @@ public class ContentHeaderBody implement
{
return new AMQFrame(channelId, body);
}
+
+ public ContentHeaderProperties getProperties()
+ {
+ return properties;
+ }
+
+ public void setProperties(ContentHeaderProperties props)
+ {
+ properties = props;
+ }
}
Modified: qpid/branches/qpid-2920/qpid/tools/src/py/qpid-config
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-2920/qpid/tools/src/py/qpid-config?rev=1087871&r1=1087870&r2=1087871&view=diff
==============================================================================
--- qpid/branches/qpid-2920/qpid/tools/src/py/qpid-config (original)
+++ qpid/branches/qpid-2920/qpid/tools/src/py/qpid-config Fri Apr 1 20:03:16 2011
@@ -173,8 +173,8 @@ def OptionsAndArguments(argv):
group3.add_option("--cluster-durable", action="store_true", help="The new queue becomes durable if there is only one functioning cluster node")
group3.add_option("--file-count", action="store", type="int", default=8, metavar="<n>", help="Number of files in queue's persistence journal")
group3.add_option("--file-size", action="store", type="int", default=24, metavar="<n>", help="File size in pages (64Kib/page)")
- group3.add_option("--max-queue-size", action="store", type="int", metavar="<n>", help="Number of files in queue's persistence journal")
- group3.add_option("--max-queue-count", action="store", type="int", metavar="<n>", help="Number of files in queue's persistence journal")
+ group3.add_option("--max-queue-size", action="store", type="int", metavar="<n>", help="Maximum in-memory queue size as bytes")
+ group3.add_option("--max-queue-count", action="store", type="int", metavar="<n>", help="Maximum in-memory queue size as a number of messages")
group3.add_option("--limit-policy", action="store", choices=["none", "reject", "flow-to-disk", "ring", "ring-strict"], metavar="<policy>", help="Action to take when queue limit is reached")
group3.add_option("--order", action="store", choices=["fifo", "lvq", "lvq-no-browse"], metavar="<ordering>", help="Queue ordering policy")
group3.add_option("--generate-queue-events", action="store", type="int", metavar="<n>", help="If set to 1, every enqueue will generate an event that can be processed by registered listeners (e.g. for replication). If set to 2, events will be generated for enqueues and dequeues.")
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org