Merge branch 'trunk' into qpid-2920

Conflicts:
	qpid/cpp/src/cluster.mk
	qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid-2920@1086439 13f79535-47bb-0310-9956-ffa450edef68

45 files changed, 902 insertions, 156 deletions

diff --git a/.gitignore b/.gitignore
index 3562b0e191..f64d5fd8e4 100644
--- a/.gitignore
+++ b/.gitignore
@@ -6,9 +6,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
@@ -35,7 +35,6 @@ autom4te.cache
 aclocal.m4
 developer.doxygen
 user.doxygen
-qpidd.1
 qpid/cpp/libtool
 qpidc.spec
 qpid/cpp/src/gen/
diff --git a/qpid/cpp/bindings/qmf/tests/test_base.rb b/qpid/cpp/bindings/qmf/tests/test_base.rb
index 3e4337a9c0..7d4609097c 100644
--- a/qpid/cpp/bindings/qmf/tests/test_base.rb
+++ b/qpid/cpp/bindings/qmf/tests/test_base.rb
@@ -24,6 +24,7 @@ require 'socket'
 
 class ConsoleTestBase < Qmf::ConsoleHandler
   def initialize
+    sleep(2)
     @settings = Qmf::ConnectionSettings.new
     @settings.host = ARGV[0] if ARGV.size > 0
     @settings.port = ARGV[1].to_i if ARGV.size > 1
@@ -67,7 +68,7 @@ class ConsoleTestBase < Qmf::ConsoleHandler
 
   def assert(condition, in_text=nil)
     text = " (#{in_text})" if in_text
-    raise "Assertion failed: #{left} != #{right}#{text}" unless condition
+    raise "Assertion failed: #{condition} #{text}" unless condition
   end
 
   def fail(text)
diff --git a/qpid/cpp/docs/man/Makefile.am b/qpid/cpp/docs/man/Makefile.am
index 14295f73bf..b821568f81 100644
--- a/qpid/cpp/docs/man/Makefile.am
+++ b/qpid/cpp/docs/man/Makefile.am
@@ -16,10 +16,29 @@
 # specific language governing permissions and limitations
 # under the License.
 #
+
+# Generate makefile from qpidd --help
+#
+# Note: qiddd.1 is normally a _checked in_ pre-generated file, so that
+# make dist does not have to build the entire source just for the man page.
+#
+# To update the checked-in file (e.g. for a new release) do the following:
+#
+# - start with a completely clean checkout.
+# - make sure there are no modules installed in your configured prefix,
+#   we don't want to pick up configuration from optional modules
+# - do bootstrap; configure
+# - in build-dir: cd src; make # build the broker
+# - in source-dir: cd docs/man; rm qpidd.1 # remove checked-in man page.
+# - in build-dir: cd docs/man; make # make new man page
+# - edit qpidd.1 to remove all default values referring to file/directory locations.
+#   these values will differ between builds depending on configuration.
+# - if source-dir != build-dir: copy qpidd.1 from build-dir/docs/man to source-dir/docs/man
+
 dist_man_MANS = qpidd.1
 
-man_aux = $(dist_man_MANS:.1=.x) 
-EXTRA_DIST = $(man_aux) generate_manpage groffify_options.sed groffify_template.sed 
+man_aux = $(dist_man_MANS:.1=.x)
+EXTRA_DIST = $(man_aux) generate_manpage groffify_options.sed groffify_template.sed
 DISTCLEANFILES = $(dist_man_MANS)
 CLEANFILES=qpidd.1
 
diff --git a/qpid/cpp/docs/man/qpidd.1 b/qpid/cpp/docs/man/qpidd.1
new file mode 100644
index 0000000000..d2cff454cf
--- /dev/null
+++ b/qpid/cpp/docs/man/qpidd.1
@@ -0,0 +1,247 @@
+.\" DO NOT MODIFY THIS FILE!  It was generated by help2man 1.38.2.
+.TH QPIDD "1" "March 2011" "qpidd (qpidc) version 0.11" "User Commands"
+.SH NAME
+
+qpidd \- the Qpid AMQP Message Broker Daemon
+.SH SYNOPSIS
+
+qpidd [-p port] [--config config_file] [--data-dir directory]
+.SH DESCRIPTION
+
+An AMQP message broker daemon that stores, routes and forwards
+messages using the Advanced Message Queueing Protocol (AMQP).
+.SH OPTIONS
+
+The options below are built-in to qpidd. Installing add-on modules provides additional options. To see the full set of options available type "qpidd --help"
+
+Options may be specified via command line, environment variable or configuration file. See FILES and ENVIRONMENT below for details.
+.PP
+
+.TP
+\fB\-h\fR [ \fB\-\-help\fR ]
+Displays the help message
+.TP
+\fB\-v\fR [ \fB\-\-version\fR ]
+Displays version information
+.TP
+\fB\-\-config\fR FILE
+Reads configuration from FILE
+.SS "Module options:"
+.TP
+\fB\-\-module\-dir\fR DIR
+Load all shareable modules in this
+directory
+.TP
+\fB\-\-load\-module\fR FILE
+Specifies additional module(s) to be
+loaded
+.TP
+\fB\-\-no\-module\-dir\fR
+Don't load modules from module
+directory
+.SS "Broker Options:"
+.TP
+\fB\-\-data\-dir\fR DIR
+Directory to contain persistent data
+generated by the broker
+.TP
+\fB\-\-no\-data\-dir\fR
+Don't use a data directory.  No
+persistent configuration will be loaded
+or stored
+.TP
+\fB\-p\fR [ \fB\-\-port\fR ] PORT (5672)
+Tells the broker to listen on PORT
+.TP
+\fB\-\-worker\-threads\fR N (3)
+Sets the broker thread pool size
+.TP
+\fB\-\-max\-connections\fR N (500)
+Sets the maximum allowed connections
+.TP
+\fB\-\-connection\-backlog\fR N (10)
+Sets the connection backlog limit for
+the server socket
+.TP
+\fB\-m\fR [ \fB\-\-mgmt\-enable\fR ] yes|no (1)
+Enable Management
+.TP
+\fB\-\-mgmt\-qmf2\fR yes|no (1)
+Enable broadcast of management
+information over QMF v2
+.TP
+\fB\-\-mgmt\-qmf1\fR yes|no (1)
+Enable broadcast of management
+information over QMF v1
+.TP
+\fB\-\-mgmt\-pub\-interval\fR SECONDS (10)
+Management Publish Interval
+.TP
+\fB\-\-queue\-purge\-interval\fR SECONDS (600)
+Interval between attempts to purge any
+expired messages from queues
+.TP
+\fB\-\-auth\fR yes|no (1)
+Enable authentication, if disabled all
+incoming connections will be trusted
+.TP
+\fB\-\-realm\fR REALM (QPID)
+Use the given realm when performing
+authentication
+.TP
+\fB\-\-default\-queue\-limit\fR BYTES (104857600)
+Default maximum size for queues (in
+bytes)
+.TP
+\fB\-\-tcp\-nodelay\fR
+Set TCP_NODELAY on TCP connections
+.TP
+\fB\-\-require\-encryption\fR
+Only accept connections that are
+encrypted
+.TP
+\fB\-\-known\-hosts\-url\fR URL or 'none'
+URL to send as 'known\-hosts' to clients
+('none' implies empty list)
+.TP
+\fB\-\-sasl\-config\fR DIR
+gets sasl config info from nonstandard
+location
+.TP
+\fB\-\-max\-session\-rate\fR MESSAGES/S (0)
+Sets the maximum message rate per
+session (0=unlimited)
+.TP
+\fB\-\-async\-queue\-events\fR yes|no (0)
+Set Queue Events async, used for
+services like replication
+.TP
+\fB\-\-default\-flow\-stop\-threshold\fR PERCENT (80)
+Percent of queue's maximum capacity at
+which flow control is activated.
+.TP
+\fB\-\-default\-flow\-resume\-threshold\fR PERCENT (70)
+Percent of queue's maximum capacity at
+which flow control is de\-activated.
+.TP
+\fB\-\-default\-event\-threshold\-ratio\fR %age of limit (80)
+The ratio of any specified queue limit
+at which an event will be raised
+.SS "Logging options:"
+.TP
+\fB\-t\fR [ \fB\-\-trace\fR ]
+Enables all logging
+.TP
+\fB\-\-log\-enable\fR RULE (notice+)
+Enables logging for selected levels and
+components. RULE is in the form
+\&'LEVEL[+][:PATTERN]' Levels are one of:
+.IP
+trace debug info notice warning error
+.IP
+critical
+For example:
+\&'\-\-log\-enable warning+' logs all
+warning, error and critical messages.
+\&'\-\-log\-enable debug:framing' logs debug
+messages from the framing namespace.
+This option can be used multiple times
+.TP
+\fB\-\-log\-time\fR yes|no (1)
+Include time in log messages
+.TP
+\fB\-\-log\-level\fR yes|no (1)
+Include severity level in log messages
+.TP
+\fB\-\-log\-source\fR yes|no (0)
+Include source file:line in log
+messages
+.TP
+\fB\-\-log\-thread\fR yes|no (0)
+Include thread ID in log messages
+.TP
+\fB\-\-log\-function\fR yes|no (0)
+Include function signature in log
+messages
+.TP
+\fB\-\-log\-prefix\fR STRING
+Prefix to append to all log messages
+.SS "Logging sink options:"
+.TP
+\fB\-\-log\-to\-stderr\fR yes|no (1)
+Send logging output to stderr
+.TP
+\fB\-\-log\-to\-stdout\fR yes|no (0)
+Send logging output to stdout
+.TP
+\fB\-\-log\-to\-file\fR FILE
+Send log output to FILE.
+.TP
+\fB\-\-log\-to\-syslog\fR yes|no (0)
+Send logging output to syslog;
+customize using \fB\-\-syslog\-name\fR and
+\fB\-\-syslog\-facility\fR
+.TP
+\fB\-\-syslog\-name\fR NAME (lt\-qpidd)
+Name to use in syslog messages
+.TP
+\fB\-\-syslog\-facility\fR LOG_XXX (LOG_DAEMON)
+Facility to use in syslog messages
+.SS "Daemon options:"
+.TP
+\fB\-d\fR [ \fB\-\-daemon\fR ]
+Run as a daemon. Logs to syslog by
+default in this mode.
+.TP
+\fB\-\-transport\fR TRANSPORT (tcp)
+The transport for which to return the
+port
+.TP
+\fB\-\-pid\-dir\fR DIR
+Directory where port\-specific PID file
+is stored
+.TP
+\fB\-w\fR [ \fB\-\-wait\fR ] SECONDS (600)
+Sets the maximum wait time to
+initialize the daemon. If the daemon
+fails to initialize, prints an error
+and returns 1
+.TP
+\fB\-c\fR [ \fB\-\-check\fR ]
+Prints the daemon's process ID to
+stdout and returns 0 if the daemon is
+running, otherwise returns 1
+.TP
+\fB\-q\fR [ \fB\-\-quit\fR ]
+Tells the daemon to shut down
+.SH ENVIRONMENT
+.I QPID_<option>
+.RS
+There is an environment variable for each option.
+.RE
+
+The environment variable is the option name in uppercase, prefixed with QPID_ and '.' or '-' are replaced with '_'. Environment settings are over-ridden by command line settings. For example:
+
+  export QPID_PORT=6000
+  export QPID_MAX_CONNECTIONS=10
+  export QPID_LOG_TO_FILE=/tmp/qpidd.log
+.SH FILES
+.I /etc/qpidd.conf
+.RS
+Default configuration file.
+.RE
+
+Configuration file settings are over-ridden by command line or environment variable settings. '--config <file>' or 'export QPID_CONFIG=<file>' specifies an alternate file.
+
+Each line is a name=value pair. Blank lines and lines beginning with # are ignored. For example:
+
+  # My qpidd configuration file.
+  port=6000
+  max-connections=10
+  log-to-file=/tmp/qpidd.log
+.SH AUTHOR
+
+The Apache Qpid Project, dev@qpid.apache.org
+.SH "REPORTING BUGS"
+
+Please report bugs to users@qpid.apache.org
diff --git a/qpid/cpp/docs/man/qpidd.x b/qpid/cpp/docs/man/qpidd.x
index af5d9628ee..0ccf3b562a 100644
--- a/qpid/cpp/docs/man/qpidd.x
+++ b/qpid/cpp/docs/man/qpidd.x
@@ -13,6 +13,8 @@ messages using the Advanced Message Queueing Protocol (AMQP).
 
 [OPTIONS]
 
+The options below are built-in to qpidd. Installing add-on modules provides additional options. To see the full set of options available type "qpidd --help"
+
 Options may be specified via command line, environment variable or configuration file. See FILES and ENVIRONMENT below for details.
 
 [FILES]
diff --git a/qpid/cpp/managementgen/qmfgen/schema.py b/qpid/cpp/managementgen/qmfgen/schema.py
index afdfe42639..59e951fb6e 100755
--- a/qpid/cpp/managementgen/qmfgen/schema.py
+++ b/qpid/cpp/managementgen/qmfgen/schema.py
@@ -1731,9 +1731,9 @@ class SchemaPackage:
                   stream.write("            qmf::SchemaProperty arg(\"%s\", %s);\n" % (arg.name, typeName))
                   if subType:
                       stream.write("            arg.setSubtype(\"%s\");\n" % subType)
-                  if stat.unit:
+                  if arg.unit:
                       stream.write("            arg.setUnit(\"%s\");\n" % arg.unit)
-                  if stat.desc:
+                  if arg.desc:
                       stream.write("            arg.setDesc(\"%s\");\n" % arg.desc)
                   stream.write("            arg.setDirection(%s);\n" % self.qmfv2Dir(arg.dir))
                   stream.write("            method.addArgument(arg);\n")
diff --git a/qpid/cpp/rubygen/amqpgen.rb b/qpid/cpp/rubygen/amqpgen.rb
index 69e65a4056..20aac35194 100755
--- a/qpid/cpp/rubygen/amqpgen.rb
+++ b/qpid/cpp/rubygen/amqpgen.rb
@@ -61,7 +61,8 @@ end
 class Module
   # Add trailing _ to avoid conflict with Object methods.
   def mangle(sym)
-    (Object.method_defined? sym) ? (sym.to_s+"_").intern : sym
+    sym =  (sym.to_s+"_").to_sym if (Object.method_defined?(sym) or sym == :type)
+    sym
   end
 
   # Add attribute reader for XML attribute.
diff --git a/qpid/cpp/src/Makefile.am b/qpid/cpp/src/Makefile.am
index d7be7121c1..820d97439c 100644
--- a/qpid/cpp/src/Makefile.am
+++ b/qpid/cpp/src/Makefile.am
@@ -127,14 +127,14 @@ qpidexec_SCRIPTS =
 qpidtestdir = $(qpidexecdir)/tests
 qpidtest_PROGRAMS =
 qpidtest_SCRIPTS =
-tmoduledir = $(libdir)/qpid/tests
-tmodule_LTLIBRARIES=
+tmoduleexecdir = $(libdir)/qpid/tests
+tmoduleexec_LTLIBRARIES=
 
 AM_CXXFLAGS += -DBOOST_FILESYSTEM_VERSION=2
 
 ## Automake macros to build libraries and executables.
-qpidd_CXXFLAGS = $(AM_CXXFLAGS) -DQPIDD_MODULE_DIR=\"$(dmoduledir)\" -DQPIDD_CONF_FILE=\"$(sysconfdir)/qpidd.conf\"
-libqpidclient_la_CXXFLAGS = $(AM_CXXFLAGS) -DQPIDC_MODULE_DIR=\"$(cmoduledir)\" -DQPIDC_CONF_FILE=\"$(confdir)/qpidc.conf\"
+qpidd_CXXFLAGS = $(AM_CXXFLAGS) -DQPIDD_MODULE_DIR=\"$(dmoduleexecdir)\" -DQPIDD_CONF_FILE=\"$(sysconfdir)/qpidd.conf\"
+libqpidclient_la_CXXFLAGS = $(AM_CXXFLAGS) -DQPIDC_MODULE_DIR=\"$(cmoduleexecdir)\" -DQPIDC_CONF_FILE=\"$(confdir)/qpidc.conf\"
 
 qpidd_LDADD =					\
   libqpidbroker.la				\
@@ -200,10 +200,10 @@ lib_LTLIBRARIES = libqpidtypes.la libqpidcommon.la libqpidbroker.la libqpidclien
 # Definitions for client and daemon plugins
 PLUGINLDFLAGS=-no-undefined -module -avoid-version
 confdir=$(sysconfdir)/qpid
-dmoduledir=$(libdir)/qpid/daemon
-cmoduledir=$(libdir)/qpid/client
-dmodule_LTLIBRARIES =
-cmodule_LTLIBRARIES =
+dmoduleexecdir=$(libdir)/qpid/daemon
+cmoduleexecdir=$(libdir)/qpid/client
+dmoduleexec_LTLIBRARIES =
+cmoduleexec_LTLIBRARIES =
 
 include cluster.mk
 include acl.mk
@@ -245,7 +245,7 @@ rdma_la_LIBADD = \
 rdma_la_LDFLAGS = $(PLUGINLDFLAGS)
 rdma_la_CXXFLAGS = \
   $(AM_CXXFLAGS) -Wno-missing-field-initializers
-dmodule_LTLIBRARIES += \
+dmoduleexec_LTLIBRARIES += \
   rdma.la
 
 rdmaconnector_la_SOURCES = \
@@ -257,7 +257,7 @@ rdmaconnector_la_LIBADD = \
 rdmaconnector_la_LDFLAGS = $(PLUGINLDFLAGS)
 rdmaconnector_la_CXXFLAGS = \
   $(AM_CXXFLAGS) -Wno-missing-field-initializers
-cmodule_LTLIBRARIES += \
+cmoduleexec_LTLIBRARIES += \
   rdmaconnector.la
 
 # RDMA test/sample programs
@@ -332,6 +332,7 @@ libqpidcommon_la_SOURCES +=			\
   qpid/Address.cpp				\
   qpid/DataDir.cpp				\
   qpid/DataDir.h				\
+  qpid/DisableExceptionLogging.h		\
   qpid/Exception.cpp				\
   qpid/Modules.cpp				\
   qpid/Modules.h				\
@@ -887,10 +888,6 @@ nobase_include_HEADERS +=			\
   ../include/qpid/types/Variant.h		\
   ../include/qpid/types/ImportExport.h
 
-# Force build of qpidd during dist phase so help2man will work.
-dist-hook: $(BUILT_SOURCES)
-	$(MAKE) qpidd
-
 # Create the default data directory
 install-data-local:
 	$(mkinstalldirs) $(DESTDIR)/$(localstatedir)/lib/qpidd
diff --git a/qpid/cpp/src/acl.mk b/qpid/cpp/src/acl.mk
index bcd1d88335..b8e2ff0e13 100644
--- a/qpid/cpp/src/acl.mk
+++ b/qpid/cpp/src/acl.mk
@@ -18,8 +18,8 @@
 #
 #
 # acl library makefile fragment, to be included in Makefile.am
-# 
-dmodule_LTLIBRARIES += acl.la
+#
+dmoduleexec_LTLIBRARIES += acl.la
 
 acl_la_SOURCES = \
   qpid/acl/Acl.cpp \
diff --git a/qpid/cpp/src/cluster.mk b/qpid/cpp/src/cluster.mk
index 86448d991c..cbb48798c1 100644
--- a/qpid/cpp/src/cluster.mk
+++ b/qpid/cpp/src/cluster.mk
@@ -34,7 +34,8 @@ endif
 
 if HAVE_LIBCPG
 
-dmodule_LTLIBRARIES += cluster.la
+dmoduleexec_LTLIBRARIES += cluster.la
+
 cluster_la_SOURCES =				\
   $(CMAN_SOURCES)				\
   qpid/cluster/Cluster.cpp			\
@@ -128,7 +129,7 @@ cluster2_la_SOURCES =				\
 
 
 # The watchdog plugin and helper executable
-dmodule_LTLIBRARIES += watchdog.la
+dmoduleexec_LTLIBRARIES += watchdog.la
 watchdog_la_SOURCES = qpid/cluster/WatchDogPlugin.cpp
 watchdog_la_LIBADD = libqpidbroker.la
 watchdog_la_LDFLAGS = $(PLUGINLDFLAGS)
diff --git a/qpid/cpp/src/qpid/DisableExceptionLogging.h b/qpid/cpp/src/qpid/DisableExceptionLogging.h
new file mode 100644
index 0000000000..04a9240513
--- /dev/null
+++ b/qpid/cpp/src/qpid/DisableExceptionLogging.h
@@ -0,0 +1,39 @@
+#ifndef QPID_DISABLEEXCEPTIONLOGGING_H
+#define QPID_DISABLEEXCEPTIONLOGGING_H
+
+/*
+ *
+ * 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.
+ *
+ */
+#include "qpid/CommonImportExport.h"
+
+namespace qpid {
+
+/**
+ * Temporarily disable logging in qpid::Exception constructor.
+ * Used by log::Logger to avoid logging exceptions during Logger construction.
+ */
+struct DisableExceptionLogging
+{
+    QPID_COMMON_EXTERN DisableExceptionLogging();
+    QPID_COMMON_EXTERN ~DisableExceptionLogging();
+};
+} // namespace qpid
+
+#endif  /*!QPID_DISABLEEXCEPTIONLOGGING_H*/
diff --git a/qpid/cpp/src/qpid/Exception.cpp b/qpid/cpp/src/qpid/Exception.cpp
index 16a3a13d17..a6696f06e1 100644
--- a/qpid/cpp/src/qpid/Exception.cpp
+++ b/qpid/cpp/src/qpid/Exception.cpp
@@ -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
@@ -21,13 +21,25 @@
 
 #include "qpid/log/Statement.h"
 #include "qpid/Exception.h"
+#include "qpid/DisableExceptionLogging.h"
 #include <typeinfo>
 #include <assert.h>
 #include <string.h>
 
 namespace qpid {
 
+// Note on static initialization order: if an exception is constructed
+// in a static constructor before disableExceptionLogging has been
+// initialized, the worst that can happen is we lose an exception log
+// message. Since we shouldn't be throwing a lot of exceptions during
+// static construction this seems safe.
+static bool disableExceptionLogging = false;
+
+DisableExceptionLogging::DisableExceptionLogging() { disableExceptionLogging = true; }
+DisableExceptionLogging::~DisableExceptionLogging() { disableExceptionLogging = false; }
+
 Exception::Exception(const std::string& msg) throw() : message(msg) {
+    if (disableExceptionLogging) return;
     QPID_LOG_IF(debug, !msg.empty(), "Exception constructed: " << message);
 }
 
diff --git a/qpid/cpp/src/qpid/acl/Acl.cpp b/qpid/cpp/src/qpid/acl/Acl.cpp
index 67603015d7..4b3dda7962 100644
--- a/qpid/cpp/src/qpid/acl/Acl.cpp
+++ b/qpid/cpp/src/qpid/acl/Acl.cpp
@@ -180,7 +180,10 @@ Acl::Acl (AclValues& av, Broker& b): aclValues(av), broker(&b), transferAcl(fals
       {
       case _qmf::Acl::METHOD_RELOADACLFILE :
           readAclFile(text);
-          status = Manageable::STATUS_USER;
+          if (text.empty())
+              status = Manageable::STATUS_OK;
+          else
+              status = Manageable::STATUS_USER;
           break;
       }
 
diff --git a/qpid/cpp/src/qpid/broker/Broker.cpp b/qpid/cpp/src/qpid/broker/Broker.cpp
index 1b64fb2a52..3edcb2c0eb 100644
--- a/qpid/cpp/src/qpid/broker/Broker.cpp
+++ b/qpid/cpp/src/qpid/broker/Broker.cpp
@@ -157,8 +157,8 @@ Broker::Options::Options(const std::string& name) :
         ("sasl-config", optValue(saslConfigPath, "DIR"), "gets sasl config info from nonstandard location")
         ("max-session-rate", optValue(maxSessionRate, "MESSAGES/S"), "Sets the maximum message rate per session (0=unlimited)")
         ("async-queue-events", optValue(asyncQueueEvents, "yes|no"), "Set Queue Events async, used for services like replication")
-        ("default-flow-stop-threshold", optValue(queueFlowStopRatio, "%MESSAGES"), "Queue capacity level at which flow control is activated.")
-        ("default-flow-resume-threshold", optValue(queueFlowResumeRatio, "%MESSAGES"), "Queue capacity level at which flow control is de-activated.")
+        ("default-flow-stop-threshold", optValue(queueFlowStopRatio, "PERCENT"), "Percent of queue's maximum capacity at which flow control is activated.")
+        ("default-flow-resume-threshold", optValue(queueFlowResumeRatio, "PERCENT"), "Percent of queue's maximum capacity at which flow control is de-activated.")
         ("default-event-threshold-ratio", optValue(queueThresholdEventRatio, "%age of limit"), "The ratio of any specified queue limit at which an event will be raised");
 }
 
diff --git a/qpid/cpp/src/qpid/log/Logger.cpp b/qpid/cpp/src/qpid/log/Logger.cpp
index 2217cdddbd..2339a62114 100644
--- a/qpid/cpp/src/qpid/log/Logger.cpp
+++ b/qpid/cpp/src/qpid/log/Logger.cpp
@@ -22,6 +22,7 @@
 #include "qpid/memory.h"
 #include "qpid/sys/Thread.h"
 #include "qpid/sys/Time.h"
+#include "qpid/DisableExceptionLogging.h"
 #include <boost/pool/detail/singleton.hpp>
 #include <boost/bind.hpp>
 #include <boost/function.hpp>
@@ -48,11 +49,16 @@ Logger& Logger::instance() {
 }
 
 Logger::Logger() : flags(0) {
+    // Disable automatic logging in Exception constructors to avoid
+    // re-entrant use of logger singleton if there is an error in
+    // option parsing.
+    DisableExceptionLogging del;
+
     // Initialize myself from env variables so all programs
     // (e.g. tests) can use logging even if they don't parse
     // command line args.
     Options opts("");
-    opts.parse(0, 0);           
+    opts.parse(0, 0);
     configure(opts);
 }
 
@@ -73,7 +79,7 @@ void Logger::log(const Statement& s, const std::string& msg) {
     std::ostringstream os;
     if (!prefix.empty())
         os << prefix << ": ";
-    if (flags&TIME) 
+    if (flags&TIME)
 		qpid::sys::outputFormattedNow(os);
     if (flags&LEVEL)
         os << LevelTraits::name(s.level) << " ";
@@ -140,7 +146,7 @@ void Logger::configure(const Options& opts) {
     Options o(opts);
     if (o.trace)
         o.selectors.push_back("trace+");
-    format(o); 
+    format(o);
     select(Selector(o));
     setPrefix(opts.prefix);
     options.sinkOptions->setup(this);
diff --git a/qpid/cpp/src/qpid/sys/Timer.cpp b/qpid/cpp/src/qpid/sys/Timer.cpp
index a97ccd1bd1..85e410860d 100644
--- a/qpid/cpp/src/qpid/sys/Timer.cpp
+++ b/qpid/cpp/src/qpid/sys/Timer.cpp
@@ -131,12 +131,14 @@ void Timer::run()
                 bool warningsEnabled;
                 QPID_LOG_TEST(warning, warningsEnabled);
                 if (warningsEnabled) {
-                    if (delay > late && overrun > overran)
-                        warn.lateAndOverran(t->name, delay, overrun, Duration(start, end));
+                    if (overrun > overran) {
+                        if (delay > overran) // if delay is significant to an overrun.
+                            warn.lateAndOverran(t->name, delay, overrun, Duration(start, end));
+                        else
+                            warn.overran(t->name, overrun, Duration(start, end));
+                    }
                     else if (delay > late)
                         warn.late(t->name, delay);
-                    else if (overrun > overran)
-                        warn.overran(t->name, overrun, Duration(start, end));
                 }
                 continue;
             } else {
diff --git a/qpid/cpp/src/qpid/sys/TimerWarnings.cpp b/qpid/cpp/src/qpid/sys/TimerWarnings.cpp
index 48a56eb472..87c3169456 100644
--- a/qpid/cpp/src/qpid/sys/TimerWarnings.cpp
+++ b/qpid/cpp/src/qpid/sys/TimerWarnings.cpp
@@ -59,17 +59,19 @@ void TimerWarnings::log() {
                 QPID_LOG(warning, task << " task late "
                          << stats.lateDelay.count << " times by "
                          << stats.lateDelay.average()/TIME_MSEC << "ms on average.");
+
             if (stats.overranOverrun.count)
                 QPID_LOG(warning, task << " task overran "
                          << stats.overranOverrun.count << " times by "
                          << stats.overranOverrun.average()/TIME_MSEC << "ms (taking "
                          << stats.overranTime.average() << "ns) on average.");
 
-            if (stats.lateAndOverranDelay.count)
-                QPID_LOG(warning, task << " task overran "
-                         << stats.overranOverrun.count << " times by "
-                         << stats.overranOverrun.average()/TIME_MSEC << "ms (taking "
-                         << stats.overranTime.average() << "ns) on average.");
+            if (stats.lateAndOverranOverrun.count)
+                QPID_LOG(warning, task << " task late and overran "
+                         << stats.lateAndOverranOverrun.count << " times: late "
+                         << stats.lateAndOverranDelay.average()/TIME_MSEC << "ms, overran "
+                         << stats.lateAndOverranOverrun.average()/TIME_MSEC << "ms (taking "
+                         << stats.lateAndOverranTime.average() << "ns) on average.");
 
         }
         nextReport = AbsTime(now(), interval);
diff --git a/qpid/cpp/src/replication.mk b/qpid/cpp/src/replication.mk
index dbe071f405..e5da32f88b 100644
--- a/qpid/cpp/src/replication.mk
+++ b/qpid/cpp/src/replication.mk
@@ -19,14 +19,14 @@
 # Make file for building two plugins for asynchronously replicating
 # queues.
 
-dmodule_LTLIBRARIES += replicating_listener.la replication_exchange.la
+dmoduleexec_LTLIBRARIES += replicating_listener.la replication_exchange.la
 
 # a queue event listener plugin that creates messages on a replication
 # queue corresponding to enqueue and dequeue events:
 replicating_listener_la_SOURCES =  \
 	qpid/replication/constants.h \
 	qpid/replication/ReplicatingEventListener.cpp \
-	qpid/replication/ReplicatingEventListener.h 
+	qpid/replication/ReplicatingEventListener.h
 
 replicating_listener_la_LIBADD = libqpidbroker.la
 if SUNOS
@@ -41,7 +41,7 @@ replicating_listener_la_LDFLAGS = $(PLUGINLDFLAGS)
 replication_exchange_la_SOURCES =  \
 	qpid/replication/constants.h \
 	qpid/replication/ReplicationExchange.cpp \
-	qpid/replication/ReplicationExchange.h 
+	qpid/replication/ReplicationExchange.h
 
 replication_exchange_la_LIBADD = libqpidbroker.la
 
diff --git a/qpid/cpp/src/ssl.mk b/qpid/cpp/src/ssl.mk
index 5fbdd55438..4dba9bb61c 100644
--- a/qpid/cpp/src/ssl.mk
+++ b/qpid/cpp/src/ssl.mk
@@ -18,7 +18,7 @@
 #
 #
 # Makefile fragment, conditionally included in Makefile.am
-# 
+#
 libsslcommon_la_SOURCES = \
   qpid/sys/ssl/check.h \
   qpid/sys/ssl/check.cpp \
@@ -47,7 +47,7 @@ ssl_la_CXXFLAGS=$(AM_CXXFLAGS) $(SSL_CFLAGS)
 
 ssl_la_LDFLAGS = $(PLUGINLDFLAGS)
 
-dmodule_LTLIBRARIES += ssl.la
+dmoduleexec_LTLIBRARIES += ssl.la
 
 sslconnector_la_SOURCES = \
   qpid/client/SslConnector.cpp
@@ -60,5 +60,5 @@ sslconnector_la_CXXFLAGS = $(AM_CXXFLAGS) -DQPIDC_CONF_FILE=\"$(confdir)/qpidc.c
 
 sslconnector_la_LDFLAGS = $(PLUGINLDFLAGS)
 
-cmodule_LTLIBRARIES += \
+cmoduleexec_LTLIBRARIES += \
   sslconnector.la
diff --git a/qpid/cpp/src/tests/brokertest.py b/qpid/cpp/src/tests/brokertest.py
index e1ee6d4a23..39eec466f2 100644
--- a/qpid/cpp/src/tests/brokertest.py
+++ b/qpid/cpp/src/tests/brokertest.py
@@ -509,6 +509,10 @@ class BrokerTest(TestCase):
         r.close()
         self.assertEqual(expect_contents, actual_contents)
 
+def join(thread, timeout=1):
+    thread.join(timeout)
+    if thread.isAlive(): raise Exception("Timed out joining thread %s"%thread)
+
 class RethrownException(Exception):
     """Captures the stack trace of the current exception to be thrown later"""
     def __init__(self, msg=""):
@@ -526,7 +530,7 @@ class StoppableThread(Thread):
 
     def stop(self):
         self.stopped = True
-        self.join()
+        join(self)
         if self.error: raise self.error
 
 class NumberedSender(Thread):
@@ -585,7 +589,7 @@ class NumberedSender(Thread):
             self.stopped = True
             self.condition.notify()
         finally: self.condition.release()
-        self.join()
+        join(self)
         self.write_message(-1)          # end-of-messages marker.
         if self.error: raise self.error
 
@@ -632,7 +636,7 @@ class NumberedReceiver(Thread):
 
     def stop(self):
         """Returns when termination message is received"""
-        self.join()
+        join(self)
         if self.error: raise self.error
 
 class ErrorGenerator(StoppableThread):
diff --git a/qpid/cpp/src/tests/sasl_fed b/qpid/cpp/src/tests/sasl_fed
index 9845a20838..884c44177c 100755
--- a/qpid/cpp/src/tests/sasl_fed
+++ b/qpid/cpp/src/tests/sasl_fed
@@ -123,7 +123,7 @@ n_messages=100
 #--------------------------------------------------
 #echo "  Sending 100 messages to $broker_1_port "
 #--------------------------------------------------
-$builddir/datagen --count $n_messages | $SENDER_EXEC --username zag --password zag --exchange $EXCHANGE_NAME --routing-key $ROUTING_KEY --port $broker_1_port
+$builddir/datagen --count $n_messages | $SENDER_EXEC --mechanism DIGEST-MD5 --username zag --password zag --exchange $EXCHANGE_NAME --routing-key $ROUTING_KEY --port $broker_1_port
 
 sleep 5
 
diff --git a/qpid/cpp/src/tests/sasl_fed_ex b/qpid/cpp/src/tests/sasl_fed_ex
index 71dece56a9..272b42cd0d 100755
--- a/qpid/cpp/src/tests/sasl_fed_ex
+++ b/qpid/cpp/src/tests/sasl_fed_ex
@@ -160,7 +160,7 @@ export QPID_SSL_CERT_NAME=${TEST_HOSTNAME}
 
 COMMON_BROKER_OPTIONS="                          \
       --ssl-sasl-no-dict                         \
-      --sasl-config=$sasl_config_dir            \
+      --sasl-config=$sasl_config_dir             \
       --ssl-require-client-authentication        \
       --auth yes                                 \
       --ssl-cert-db $CERT_DIR                    \
@@ -169,7 +169,6 @@ COMMON_BROKER_OPTIONS="                          \
       --no-data-dir                              \
       --no-module-dir                            \
       --load-module ${SSL_LIB}                   \
-      --load-module ${CLUSTER_LIB}               \
       --mgmt-enable=yes                          \
       --log-enable info+                         \
       --log-source yes                           \
@@ -186,6 +185,7 @@ function start_brokers {
       --port=${SRC_TCP_PORT}                     \
       --ssl-port ${SRC_SSL_PORT}                 \
       ${COMMON_BROKER_OPTIONS}                   \
+      --load-module ${CLUSTER_LIB}               \
       --cluster-name ${CLUSTER_1_NAME}           \
       --log-to-file $tmp_root/qpidd_src.log 2> /dev/null
 
@@ -196,6 +196,7 @@ function start_brokers {
       --port=${SRC_TCP_PORT_2}                   \
       --ssl-port ${SRC_SSL_PORT_2}               \
       ${COMMON_BROKER_OPTIONS}                   \
+      --load-module ${CLUSTER_LIB}               \
       --cluster-name ${CLUSTER_1_NAME}           \
       --log-to-file $tmp_root/qpidd_src_2.log 2> /dev/null
 
@@ -209,6 +210,7 @@ function start_brokers {
       --port=${DST_TCP_PORT}                     \
       --ssl-port ${DST_SSL_PORT}                 \
       ${COMMON_BROKER_OPTIONS}                   \
+      --load-module ${CLUSTER_LIB}               \
       --cluster-name ${CLUSTER_2_NAME}           \
       --log-to-file $tmp_root/qpidd_dst.log 2> /dev/null
 
@@ -219,6 +221,7 @@ function start_brokers {
       --port=${DST_TCP_PORT_2}                   \
       --ssl-port ${DST_SSL_PORT_2}               \
       ${COMMON_BROKER_OPTIONS}                   \
+      --load-module ${CLUSTER_LIB}               \
       --cluster-name ${CLUSTER_2_NAME}           \
       --log-to-file $tmp_root/qpidd_dst_2.log 2> /dev/null
 
@@ -329,12 +332,10 @@ halt_brokers
 
 sleep 1
 
-print "Removing temporary directory $tmp_root"
-rm -rf $tmp_root
-
-
 if [ ${link_status} == "Operational" ]; then
   print "result: good"
+  print "Removing temporary directory $tmp_root"
+  rm -rf $tmp_root
   exit 0
 fi
 
diff --git a/qpid/cpp/src/tests/sasl_fed_ex_dynamic_cluster b/qpid/cpp/src/tests/sasl_fed_ex_dynamic_cluster
index 273a543ec5..b0cceccecb 100755
--- a/qpid/cpp/src/tests/sasl_fed_ex_dynamic_cluster
+++ b/qpid/cpp/src/tests/sasl_fed_ex_dynamic_cluster
@@ -21,7 +21,8 @@
 
 
 source ./test_env.sh
+source $srcdir/ais_check
 
-${srcdir}/sasl_fed_ex dynamic cluster
+with_ais_group ${srcdir}/sasl_fed_ex dynamic cluster
 
 
diff --git a/qpid/cpp/src/tests/sasl_fed_ex_link_cluster b/qpid/cpp/src/tests/sasl_fed_ex_link_cluster
index 0e037b1969..4139300b12 100755
--- a/qpid/cpp/src/tests/sasl_fed_ex_link_cluster
+++ b/qpid/cpp/src/tests/sasl_fed_ex_link_cluster
@@ -21,7 +21,8 @@
 
 
 source ./test_env.sh
+source $srcdir/ais_check
 
-${srcdir}/sasl_fed_ex link cluster
+with_ais_group ${srcdir}/sasl_fed_ex link cluster
 
 
diff --git a/qpid/cpp/src/tests/sasl_fed_ex_queue_cluster b/qpid/cpp/src/tests/sasl_fed_ex_queue_cluster
index ec8750b861..f251420e08 100755
--- a/qpid/cpp/src/tests/sasl_fed_ex_queue_cluster
+++ b/qpid/cpp/src/tests/sasl_fed_ex_queue_cluster
@@ -21,7 +21,8 @@
 
 
 source ./test_env.sh
+source ${srcdir}/ais_check
 
-${srcdir}/sasl_fed_ex queue cluster
+with_ais_group ${srcdir}/sasl_fed_ex queue cluster
 
 
diff --git a/qpid/cpp/src/tests/sasl_fed_ex_route_cluster b/qpid/cpp/src/tests/sasl_fed_ex_route_cluster
index fd59680b6e..a5d1542def 100755
--- a/qpid/cpp/src/tests/sasl_fed_ex_route_cluster
+++ b/qpid/cpp/src/tests/sasl_fed_ex_route_cluster
@@ -21,7 +21,8 @@
 
 
 source ./test_env.sh
+source ${srcdir}/ais_check
 
-${srcdir}/sasl_fed_ex route cluster
+with_ais_group ${srcdir}/sasl_fed_ex route cluster
 
 
diff --git a/qpid/cpp/src/xml.mk b/qpid/cpp/src/xml.mk
index 0d700fcc03..baf3803647 100644
--- a/qpid/cpp/src/xml.mk
+++ b/qpid/cpp/src/xml.mk
@@ -16,7 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-dmodule_LTLIBRARIES += xml.la
+dmoduleexec_LTLIBRARIES += xml.la
 
 xml_la_SOURCES =  \
 	qpid/xml/XmlExchange.cpp \
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
index b003152db6..0e3f7b2625 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
@@ -1808,14 +1808,40 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
     }
 
 
+    /**
+     * Used by queue Runners to asynchronously deliver messages to consumers.
+     *
+     * A queue Runner is started whenever a state change occurs, e.g when a new
+     * message arrives on the queue and cannot be immediately delivered to a
+     * subscription (i.e. asynchronous delivery is required). Unless there are
+     * SubFlushRunners operating (due to subscriptions unsuspending) which are
+     * capable of accepting/delivering all messages then these messages would
+     * otherwise remain on the queue.
+     *
+     * processQueue should be running while there are messages on the queue AND
+     * there are subscriptions that can deliver them. If there are no
+     * subscriptions capable of delivering the remaining messages on the queue
+     * then processQueue should stop to prevent spinning.
+     *
+     * Since processQueue is runs in a fixed size Executor, it should not run
+     * indefinitely to prevent starving other tasks of CPU (e.g jobs to process
+     * incoming messages may not be able to be scheduled in the thread pool
+     * because all threads are working on clearing down large queues). To solve
+     * this problem, after an arbitrary number of message deliveries the
+     * processQueue job stops iterating, resubmits itself to the executor, and
+     * ends the current instance
+     *
+     * @param runner the Runner to schedule
+     * @throws AMQException
+     */
     private void processQueue(Runnable runner) throws AMQException
     {
         long stateChangeCount;
         long previousStateChangeCount = Long.MIN_VALUE;
         boolean deliveryIncomplete = true;
 
-        int extraLoops = 1;
-        long iterations = MAX_ASYNC_DELIVERIES;
+        boolean lastLoop = false;
+        int iterations = MAX_ASYNC_DELIVERIES;
 
         _asynchronousRunner.compareAndSet(runner, null);
 
@@ -1832,12 +1858,14 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
 
             if (previousStateChangeCount != stateChangeCount)
             {
-                extraLoops = 1;
+                //further asynchronous delivery is required since the
+                //previous loop. keep going if iteration slicing allows.
+                lastLoop = false;
             }
 
             previousStateChangeCount = stateChangeCount;
-            deliveryIncomplete = _subscriptionList.size() != 0;
-            boolean done;
+            boolean allSubscriptionsDone = true;
+            boolean subscriptionDone;
 
             SubscriptionList.SubscriptionNodeIterator subscriptionIter = _subscriptionList.iterator();
             //iterate over the subscribers and try to advance their pointer
@@ -1847,30 +1875,25 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
                 sub.getSendLock();
                 try
                 {
-
-                    done = attemptDelivery(sub);
-
-                    if (done)
+                    //attempt delivery. returns true if no further delivery currently possible to this sub
+                    subscriptionDone = attemptDelivery(sub);
+                    if (subscriptionDone)
                     {
-                        if (extraLoops == 0)
+                        //close autoClose subscriptions if we are not currently intent on continuing
+                        if (lastLoop && sub.isAutoClose())
                         {
-                            deliveryIncomplete = false;
-                            if (sub.isAutoClose())
-                            {
-                                unregisterSubscription(sub);
+                            unregisterSubscription(sub);
 
-                                sub.confirmAutoClose();
-                            }
-                        }
-                        else
-                        {
-                            extraLoops--;
+                            sub.confirmAutoClose();
                         }
                     }
                     else
                     {
+                        //this subscription can accept additional deliveries, so we must 
+                        //keep going after this (if iteration slicing allows it)
+                        allSubscriptionsDone = false;
+                        lastLoop = false;
                         iterations--;
-                        extraLoops = 1;
                     }
                 }
                 finally
@@ -1878,10 +1901,34 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
                     sub.releaseSendLock();
                 }
             }
+
+            if(allSubscriptionsDone && lastLoop)
+            {
+                //We have done an extra loop already and there are again
+                //again no further delivery attempts possible, only
+                //keep going if state change demands it.
+                deliveryIncomplete = false;
+            }
+            else if(allSubscriptionsDone)
+            {
+                //All subscriptions reported being done, but we have to do
+                //an extra loop if the iterations are not exhausted and
+                //there is still any work to be done
+                deliveryIncomplete = _subscriptionList.size() != 0;
+                lastLoop = true;
+            }
+            else
+            {
+                //some subscriptions can still accept more messages,
+                //keep going if iteration count allows.
+                lastLoop = false;
+                deliveryIncomplete = true;
+            }
+
             _asynchronousRunner.set(null);
         }
 
-        // If deliveries == 0 then the limitting factor was the time-slicing rather than available messages or credit
+        // If iterations == 0 then the limiting factor was the time-slicing rather than available messages or credit
         // therefore we should schedule this runner again (unless someone beats us to it :-) ).
         if (iterations == 0 && _asynchronousRunner.compareAndSet(null, runner))
         {
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
index 75bd50e3a2..54cd709af3 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
@@ -88,8 +88,18 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
                 _onOpenTask.run();    
             }
             _actor.message(ConnectionMessages.OPEN(getClientId(), "0-10", true, true));
+
+            getVirtualHost().getConnectionRegistry().registerConnection(this);
         }
-        
+
+        if (state == State.CLOSE_RCVD || state == State.CLOSED || state == State.CLOSING)
+        {
+            if(_virtualHost != null)
+            {
+                _virtualHost.getConnectionRegistry().deregisterConnection(this);
+            }
+        }
+
         if (state == State.CLOSED)
         {
             logClosed();
@@ -126,7 +136,6 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
     public void setVirtualHost(VirtualHost virtualHost)
     {
         _virtualHost = virtualHost;
-        _virtualHost.getConnectionRegistry().registerConnection(this);
         
         initialiseStatistics();
     }
@@ -253,7 +262,6 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
             // Ignore
         }
         close(replyCode, message);
-        getVirtualHost().getConnectionRegistry().deregisterConnection(this);
     }
 
     @Override
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
index 5fc8d43e94..d518d7b2c4 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
@@ -3461,4 +3461,9 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
     {
         return _closing.get()|| _connection.isClosing();
     }
+    
+    public boolean isDeclareExchanges()
+    {
+    	return DECLARE_EXCHANGES;
+    }
 }
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
index 36f2dfb66f..62d1d1698c 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
@@ -72,12 +72,15 @@ public class BasicMessageProducer_0_10 extends BasicMessageProducer
     {
         if (destination.getDestSyntax() == DestSyntax.BURL)
         {
-            String name = destination.getExchangeName().toString();
-            ((AMQSession_0_10) getSession()).getQpidSession().exchangeDeclare
-                (name,
-                 destination.getExchangeClass().toString(),
-                 null, null,
-                 name.startsWith("amq.") ? Option.PASSIVE : Option.NONE);
+        	if (getSession().isDeclareExchanges())
+        	{
+	            String name = destination.getExchangeName().toString();
+	            ((AMQSession_0_10) getSession()).getQpidSession().exchangeDeclare
+	                (name,
+	                 destination.getExchangeClass().toString(),
+	                 null, null,
+	                 name.startsWith("amq.") ? Option.PASSIVE : Option.NONE);
+        	}
         }
         else
         {       
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java b/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java
index e454a8eee4..a53558757e 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java
@@ -232,14 +232,9 @@ public class AddressHelper
     
     private boolean getDurability(Map map)
     {
-        if (map != null && map.get(DURABLE) != null)
-        {
-            return Boolean.parseBoolean((String)map.get(DURABLE));
-        }
-        else
-        {
-            return false;
-        }
+        Accessor access = new MapAccessor(map);
+        Boolean result = access.getBoolean(DURABLE);
+        return (result == null) ? false : result.booleanValue();
     }
 
     /**
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java b/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java
index 516204fbd3..1a57af9bf7 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java
@@ -339,7 +339,7 @@ public class FileUtils
         }
 
         //else we have a source directory
-        if (!dst.isDirectory() && !dst.mkdir())
+        if (!dst.isDirectory() && !dst.mkdirs())
         {
             throw new UnableToCopyException("Unable to create destination directory");
         }
diff --git a/qpid/java/module.xml b/qpid/java/module.xml
index 80f577b018..6ff807d377 100644
--- a/qpid/java/module.xml
+++ b/qpid/java/module.xml
@@ -323,6 +323,8 @@
   </condition>
 
   <property name="jvm.args" value=""/>
+  <property name="broker.existing.qpid.work" value=""/>
+
   <target name="test" depends="build,compile-tests" if="module.test.src.exists"
          unless="${dontruntest}" description="execute unit tests">
 
@@ -346,6 +348,7 @@
       <sysproperty key="broker" value="${broker}"/>
       <sysproperty key="broker.clean" value="${broker.clean}"/>
       <sysproperty key="broker.clean.between.tests" value="${broker.clean.between.tests}"/>
+      <sysproperty key="broker.existing.qpid.work" value="${broker.existing.qpid.work}"/>
       <sysproperty key="broker.persistent" value="${broker.persistent}"/>
       <sysproperty key="broker.version" value="${broker.version}"/>
       <sysproperty key="broker.ready" value="${broker.ready}" />
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MultipleTransactedBatchProducerTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MultipleTransactedBatchProducerTest.java
new file mode 100644
index 0000000000..460270e188
--- /dev/null
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MultipleTransactedBatchProducerTest.java
@@ -0,0 +1,246 @@
+/*
+ *
+ * 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 java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.Session;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+public class MultipleTransactedBatchProducerTest extends QpidBrokerTestCase
+{
+    private static final Logger _logger = Logger.getLogger(MultipleTransactedBatchProducerTest.class);
+
+    private static final int MESSAGE_COUNT = 1000;
+    private static final int BATCH_SIZE = 50;
+    private static final int NUM_PRODUCERS = 2;
+    private static final int NUM_CONSUMERS = 3;
+    private static final Random RANDOM = new Random();
+
+    private CountDownLatch _receivedLatch;
+    private String _queueName;
+
+    private volatile String _failMsg;
+
+    public void setUp() throws Exception
+    {
+        //debug level logging often makes this test pass artificially, turn the level down to info.
+        setSystemProperty("amqj.server.logging.level", "INFO");
+        _receivedLatch = new CountDownLatch(MESSAGE_COUNT * NUM_PRODUCERS);
+        setConfigurationProperty("management.enabled", "true");
+        super.setUp();
+        _queueName = getTestQueueName();
+        _failMsg = null;
+    }
+
+    /**
+     * When there are multiple producers submitting batches of messages to a given
+     * queue using transacted sessions, it is highly probable that concurrent
+     * enqueue() activity will occur and attempt delivery of their message to the
+     * same subscription. In this scenario it is likely that one of the attempts
+     * will succeed and the other will result in use of the deliverAsync() method
+     * to start a queue Runner and ensure delivery of the message.
+     *
+     * A defect within the processQueue() method used by the Runner 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 and found to be
+     * most visible when Selectors are used such that one and only one subscription
+     * can/will accept any given message, 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,
+     * by utilising multiple batch transacted producers to create the scenario and
+     * ensure all messages are received by a consumer.
+     */
+    public void testMultipleBatchedProducersWithMultipleConsumersUsingSelectors() throws Exception
+    {
+        String selector1 = ("(\"" + _queueName +"\" % " + NUM_CONSUMERS + ") = 0");
+        String selector2 = ("(\"" + _queueName +"\" % " + NUM_CONSUMERS + ") = 1");
+        String selector3 = ("(\"" + _queueName +"\" % " + NUM_CONSUMERS + ") = 2");
+
+        //create consumers
+        Connection conn1 = getConnection();
+        conn1.setExceptionListener(new ExceptionHandler("conn1"));
+        Session sess1 = conn1.createSession(true, Session.SESSION_TRANSACTED);
+        MessageConsumer cons1 = sess1.createConsumer(sess1.createQueue(_queueName), selector1);
+        cons1.setMessageListener(new Cons(sess1,"consumer1"));
+
+        Connection conn2 = getConnection();
+        conn2.setExceptionListener(new ExceptionHandler("conn2"));
+        Session sess2 = conn2.createSession(true, Session.SESSION_TRANSACTED);
+        MessageConsumer cons2 = sess2.createConsumer(sess2.createQueue(_queueName), selector2);
+        cons2.setMessageListener(new Cons(sess2,"consumer2"));
+
+        Connection conn3 = getConnection();
+        conn3.setExceptionListener(new ExceptionHandler("conn3"));
+        Session sess3 = conn3.createSession(true, Session.SESSION_TRANSACTED);
+        MessageConsumer cons3 = sess3.createConsumer(sess3.createQueue(_queueName), selector3);
+        cons3.setMessageListener(new Cons(sess3,"consumer3"));
+
+        conn1.start();
+        conn2.start();
+        conn3.start();
+
+        //create producers
+        Connection connA = getConnection();
+        connA.setExceptionListener(new ExceptionHandler("connA"));
+        Connection connB = getConnection();
+        connB.setExceptionListener(new ExceptionHandler("connB"));
+        Thread producer1 = new Thread(new ProducerThread(connA, _queueName, "producer1"));
+        Thread producer2 = new Thread(new ProducerThread(connB, _queueName, "producer2"));
+
+        producer1.start();
+        Thread.sleep(10);
+        producer2.start();
+
+        //await delivery of the messages
+        boolean result = _receivedLatch.await(75, TimeUnit.SECONDS);
+
+        assertNull("Test failed because: " + String.valueOf(_failMsg), _failMsg);
+        assertTrue("Some of the messages were not all recieved in the given timeframe, remaining count was: "+_receivedLatch.getCount(),
+                   result);
+
+    }
+
+    @Override
+    public Message createNextMessage(Session session, int msgCount) throws JMSException
+    {
+        Message message = super.createNextMessage(session,msgCount);
+
+        //bias at least 50% of the messages to the first consumers selector because
+        //the issue presents itself primarily when an earlier subscription completes
+        //delivery after the later subscriptions
+        int val;
+        if (msgCount % 2 == 0)
+        {
+            val = 0;
+        }
+        else
+        {
+            val = RANDOM.nextInt(Integer.MAX_VALUE);
+        }
+
+        message.setIntProperty(_queueName, val);
+
+        return message;
+    }
+
+    private class Cons implements MessageListener
+    {
+        private Session _sess;
+        private String _desc;
+
+        public Cons(Session sess, String desc)
+        {
+            _sess = sess;
+            _desc = desc;
+        }
+
+        public void onMessage(Message message)
+        {
+            _receivedLatch.countDown();
+            int msgCount = 0;
+            int msgID = 0;
+            try
+            {
+                msgCount = message.getIntProperty(INDEX);
+                msgID = message.getIntProperty(_queueName);
+            }
+            catch (JMSException e)
+            {
+                _logger.error(_desc + " received exception: " + e.getMessage(), e);
+                failAsyncTest(e.getMessage());
+            }
+
+            _logger.info("Consumer received message:"+ msgCount + " with ID: " + msgID);
+
+            try
+            {
+                _sess.commit();
+            }
+            catch (JMSException e)
+            {
+                _logger.error(_desc + " received exception: " + e.getMessage(), e);
+                failAsyncTest(e.getMessage());
+            }
+        }
+    }
+
+    private class ProducerThread implements Runnable
+    {
+        private Connection _conn;
+        private String _dest;
+        private String _desc;
+
+        public ProducerThread(Connection conn, String dest, String desc)
+        {
+            _conn = conn;
+            _dest = dest;
+            _desc = desc;
+        }
+
+        public void run()
+        {
+            try
+            {
+                Session session = _conn.createSession(true, Session.SESSION_TRANSACTED);
+                sendMessage(session, session.createQueue(_dest), MESSAGE_COUNT, BATCH_SIZE);
+            }
+            catch (Exception e)
+            {
+                _logger.error(_desc + " received exception: " + e.getMessage(), e);
+                failAsyncTest(e.getMessage());
+            }
+        }
+    }
+
+    private class ExceptionHandler implements javax.jms.ExceptionListener
+    {
+        private String _desc;
+
+        public ExceptionHandler(String description)
+        {
+            _desc = description;
+        }
+
+        public void onException(JMSException e)
+        {
+            _logger.error(_desc + " received exception: " + e.getMessage(), e);
+            failAsyncTest(e.getMessage());
+        }
+    }
+
+    private void failAsyncTest(String msg)
+    {
+        _logger.error("Failing test because: " + msg);
+        _failMsg = msg;
+    }
+}
\ No newline at end of file
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
index 30dc30cd81..a737781e30 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
@@ -474,39 +474,6 @@ public class AddressBasedDestinationTest extends QpidBrokerTestCase
     }
 
     /**
-    * Test goal: Verifies that and address based destination can be used successfully
-    *            as a reply to.
-    */
-    public void testAddressBasedReplyTo() throws Exception
-    {
-        Session jmsSession = _connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
-
-        String addr = "ADDR:amq.direct/x512; {create: receiver, " +
-                      "link : {name : 'MY.RESP.QUEUE', " +
-                      "x-declare : { auto-delete: true, exclusive: true, " +
-                                   "arguments : {'qpid.max_size': 1000, 'qpid.policy_type': ring }} } }";
-
-        Destination replyTo = new AMQAnyDestination(addr);
-        Destination dest =new AMQAnyDestination("ADDR:amq.direct/Hello");
-
-        MessageConsumer cons = jmsSession.createConsumer(dest);
-        MessageProducer prod = jmsSession.createProducer(dest);
-        Message m = jmsSession.createTextMessage("Hello");
-        m.setJMSReplyTo(replyTo);
-        prod.send(m);
-
-        Message msg = cons.receive(1000);
-        assertNotNull("consumer should have received the message",msg);
-
-        MessageConsumer replyToCons = jmsSession.createConsumer(replyTo);
-        MessageProducer replyToProd = jmsSession.createProducer(msg.getJMSReplyTo());
-        replyToProd.send(jmsSession.createTextMessage("reply"));
-
-        Message replyToMsg = replyToCons.receive(1000);
-        assertNotNull("The reply to consumer should have got the message",replyToMsg);
-    }
-
-    /**
      * Test goal: Verifies that session.createQueue method
      *            works as expected both with the new and old addressing scheme.
      */
@@ -1020,4 +987,38 @@ public class AddressBasedDestinationTest extends QpidBrokerTestCase
         prod.close();
         cons.close();
     }
+
+    public void testReplyToWithNamelessExchange() throws Exception
+    {
+    	System.setProperty("qpid.declare_exchanges","false");
+    	replyToTest("ADDR:my-queue;{create: always}");
+    	System.setProperty("qpid.declare_exchanges","true");
+    }
+
+    public void testReplyToWithCustomExchange() throws Exception
+    {
+    	replyToTest("ADDR:hello;{create:always,node:{type:topic}}");
+    }
+
+    private void replyToTest(String replyTo) throws Exception
+    {
+		Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+		Destination replyToDest = AMQDestination.createDestination(replyTo);
+	    MessageConsumer replyToCons = session.createConsumer(replyToDest);
+
+		Destination dest = session.createQueue("amq.direct/test");
+
+		MessageConsumer cons = session.createConsumer(dest);
+		MessageProducer prod = session.createProducer(dest);
+		Message m = session.createTextMessage("test");
+		m.setJMSReplyTo(replyToDest);
+		prod.send(m);
+
+		Message msg = cons.receive();
+		MessageProducer prodR = session.createProducer(msg.getJMSReplyTo());
+		prodR.send(session.createTextMessage("x"));
+
+		Message m1 = replyToCons.receive();
+		assertNotNull("The reply to consumer should have received the messsage",m1);
+    }
 }
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java
index 830421a01f..8caeaa55c0 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java
@@ -31,6 +31,7 @@ import org.apache.qpid.test.utils.QpidBrokerTestCase;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.jms.Connection;
 import javax.jms.Destination;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
@@ -39,7 +40,11 @@ import javax.jms.MessageProducer;
 import javax.jms.ObjectMessage;
 import javax.jms.Queue;
 import javax.jms.Session;
+import javax.jms.Topic;
+
 import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
 
 /**
  * @author Apache Software Foundation
@@ -163,4 +168,37 @@ public class JMSPropertiesTest extends QpidBrokerTestCase
         con.close();
     }
 
+    /**
+     * Test Goal : test if the message properties can be retrieved properly with out an error
+     *             and also test if unsupported properties are filtered out. See QPID-2930.
+     */
+    public void testGetPropertyNames() throws Exception
+    {
+        Connection con = getConnection("guest", "guest");
+        Session ssn = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        con.start();
+        
+        Topic topic = ssn.createTopic("test");
+        MessageConsumer consumer = ssn.createConsumer(topic);
+        MessageProducer prod = ssn.createProducer(topic);
+        Message m = ssn.createMessage();
+        m.setObjectProperty("x-amqp-0-10.routing-key", "routing-key".getBytes());
+        m.setObjectProperty("routing-key", "routing-key");
+        prod.send(m);
+        
+        Message msg = consumer.receive(1000);
+        assertNotNull(msg);
+        
+    	Enumeration<String> enu = msg.getPropertyNames();
+    	Map<String,String> map = new HashMap<String,String>();
+    	while (enu.hasMoreElements()) 
+    	{    		
+    		String name = enu.nextElement(); 
+    		String value = msg.getStringProperty(name);
+    		map.put(name, value);
+       } 
+    	
+       assertFalse("Property 'x-amqp-0-10.routing-key' should have been filtered out",map.containsKey("x-amqp-0-10.routing-key"));
+       assertTrue("Property routing-key should be present",map.containsKey("routing-key"));      
+    }
 }
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
index ae38a75e7a..d7da5d8a78 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
@@ -67,6 +67,7 @@ import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry;
 import org.apache.qpid.server.store.DerbyMessageStore;
 import org.apache.qpid.url.URLSyntaxException;
+import org.apache.qpid.util.FileUtils;
 import org.apache.qpid.util.LogMonitor;
 
 /**
@@ -109,6 +110,7 @@ public class QpidBrokerTestCase extends QpidTestCase
     private static final String BROKER = "broker";
     private static final String BROKER_CLEAN = "broker.clean";
     private static final String BROKER_CLEAN_BETWEEN_TESTS = "broker.clean.between.tests";
+    private static final String BROKER_EXISTING_QPID_WORK = "broker.existing.qpid.work";
     private static final String BROKER_VERSION = "broker.version";
     protected static final String BROKER_READY = "broker.ready";
     private static final String BROKER_STOPPED = "broker.stopped";
@@ -283,6 +285,19 @@ public class QpidBrokerTestCase extends QpidTestCase
             fail("Unable to test without config file:" + _configFile);
         }
 
+        if(_brokerCleanBetweenTests)
+        {
+            cleanBroker();
+
+            String existingQpidWorkPath = System.getProperty(BROKER_EXISTING_QPID_WORK);
+            if(existingQpidWorkPath != null && !existingQpidWorkPath.equals(""))
+            {
+                File existing = new File(existingQpidWorkPath);
+                File qpidWork = new File(getQpidWork(_broker, getPort()));
+                FileUtils.copyRecursive(existing, qpidWork);
+            }
+        }
+
         startBroker();
     }
 
@@ -490,7 +505,7 @@ public class QpidBrokerTestCase extends QpidTestCase
             // DON'T change PNAME, qpid.stop needs this value.
             env.put("QPID_PNAME", "-DPNAME=QPBRKR -DTNAME=\"" + _testName + "\"");
             // Add the port to QPID_WORK to ensure unique working dirs for multi broker tests
-            env.put("QPID_WORK", System.getProperty("QPID_WORK")+ "/" + port);
+            env.put("QPID_WORK", getQpidWork(_broker, port));
 
 
             // Use the environment variable to set amqj.logging.level for the broker
@@ -578,6 +593,20 @@ public class QpidBrokerTestCase extends QpidTestCase
         _brokers.put(port, process);
     }
 
+    private String getQpidWork(String broker, int port)
+    {
+        if (broker.equals(VM))
+        {
+            return System.getProperty("QPID_WORK");
+        }
+        else if (!broker.equals(EXTERNAL))
+        {
+            return System.getProperty("QPID_WORK")+ "/" + port;
+        }
+
+        return System.getProperty("QPID_WORK");
+    }
+
     public String getTestConfigFile()
     {
         String path = _output == null ? System.getProperty("java.io.tmpdir") : _output;
@@ -1190,7 +1219,8 @@ public class QpidBrokerTestCase extends QpidTestCase
 
         MessageProducer producer = session.createProducer(destination);
 
-        for (int i = offset; i < (count + offset); i++)
+        int i = offset;
+        for (; i < (count + offset); i++)
         {
             Message next = createNextMessage(session, i);
 
@@ -1213,7 +1243,7 @@ public class QpidBrokerTestCase extends QpidTestCase
         // we have no batchSize or
         // our count is not divible by batchSize.
         if (session.getTransacted() &&
-            ( batchSize == 0 || count % batchSize != 0))
+            ( batchSize == 0 || (i-1) % batchSize != 0))
         {
             session.commit();
         }
diff --git a/qpid/java/test-profiles/08StandaloneExcludes b/qpid/java/test-profiles/08StandaloneExcludes
index b482a14c6d..43eb1f8ee5 100644
--- a/qpid/java/test-profiles/08StandaloneExcludes
+++ b/qpid/java/test-profiles/08StandaloneExcludes
@@ -37,3 +37,5 @@ org.apache.qpid.test.unit.message.UTF8Test#*
 org.apache.qpid.client.MessageListenerTest#testSynchronousReceiveNoWait
 
 org.apache.qpid.test.unit.client.connection.ConnectionTest#testUnsupportedSASLMechanism
+
+org.apache.qpid.test.unit.message.JMSPropertiesTest#testGetPropertyNames
diff --git a/qpid/java/test-profiles/JavaInVMExcludes b/qpid/java/test-profiles/JavaInVMExcludes
index 7e7da4302e..7960b28d81 100644
--- a/qpid/java/test-profiles/JavaInVMExcludes
+++ b/qpid/java/test-profiles/JavaInVMExcludes
@@ -41,3 +41,5 @@ org.apache.qpid.test.unit.ack.RecoverTest#testOderingWithSyncConsumer
 
 //The VM broker does not export the logging management JMX MBean
 org.apache.qpid.server.security.acl.ExternalAdminACLTest#*
+
+org.apache.qpid.test.unit.message.JMSPropertiesTest#testGetPropertyNames
diff --git a/qpid/python/qpid/messaging/driver.py b/qpid/python/qpid/messaging/driver.py
index 2eb2c1863e..78af2827df 100644
--- a/qpid/python/qpid/messaging/driver.py
+++ b/qpid/python/qpid/messaging/driver.py
@@ -66,7 +66,7 @@ class Attachment:
 
 # XXX
 
-DURABLE_DEFAULT=True
+DURABLE_DEFAULT=False
 
 # XXX
 
@@ -526,7 +526,7 @@ class Driver:
       rawlog.debug("OPEN[%s]: %s:%s", self.log_id, host, port)
       trans = transports.TRANSPORTS.get(self.connection.transport)
       if trans:
-        self._transport = trans(host, port)
+        self._transport = trans(self.connection, host, port)
       else:
         raise ConnectError("no such transport: %s" % self.connection.transport)
       if self._retrying and self._reconnect_log:
@@ -930,6 +930,7 @@ class Engine:
 
   def resolve_declare(self, sst, lnk, dir, action):
     declare = lnk.options.get("create") in ("always", dir)
+    assrt = lnk.options.get("assert") in ("always", dir)
     def do_resolved(type, subtype):
       err = None
       if type is None:
@@ -938,7 +939,12 @@ class Engine:
         else:
           err = NotFound(text="no such queue: %s" % lnk.name)
       else:
-        action(type, subtype)
+        if assrt:
+          expected = lnk.options.get("node", {}).get("type")
+          if expected and type != expected:
+            err = AssertionFailed(text="expected %s, got %s" % (expected, type))
+        if err is None:
+          action(type, subtype)
 
       if err:
         tgt = lnk.target
diff --git a/qpid/python/qpid/messaging/endpoints.py b/qpid/python/qpid/messaging/endpoints.py
index 30c5850397..cfc89d4e2b 100644
--- a/qpid/python/qpid/messaging/endpoints.py
+++ b/qpid/python/qpid/messaging/endpoints.py
@@ -158,6 +158,7 @@ class Connection(Endpoint):
     self.reconnect_log = options.get("reconnect_log", True)
 
     self.address_ttl = options.get("address_ttl", 60)
+    self.tcp_nodelay = options.get("tcp_nodelay", False)
 
     self.options = options
 
diff --git a/qpid/python/qpid/messaging/transports.py b/qpid/python/qpid/messaging/transports.py
index 8133a45604..7abaae12e8 100644
--- a/qpid/python/qpid/messaging/transports.py
+++ b/qpid/python/qpid/messaging/transports.py
@@ -17,18 +17,23 @@
 # under the License.
 #
 
+import socket
 from qpid.util import connect
 
 TRANSPORTS = {}
 
-class tcp:
+class SocketTransport:
 
-  def __init__(self, host, port):
+  def __init__(self, conn, host, port):
     self.socket = connect(host, port)
+    if conn.tcp_nodelay:
+      self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 
   def fileno(self):
     return self.socket.fileno()
 
+class tcp(SocketTransport):
+
   def reading(self, reading):
     return reading
 
@@ -52,17 +57,14 @@ try:
 except ImportError:
   pass
 else:
-  class tls:
+  class tls(SocketTransport):
 
-    def __init__(self, host, port):
-      self.socket = connect(host, port)
+    def __init__(self, conn, host, port):
+      SocketTransport.__init__(self, conn, host, port)
       self.tls = wrap_socket(self.socket)
       self.socket.setblocking(0)
       self.state = None
 
-    def fileno(self):
-      return self.socket.fileno()
-
     def reading(self, reading):
       if self.state is None:
         return reading
diff --git a/qpid/python/qpid/tests/messaging/endpoints.py b/qpid/python/qpid/tests/messaging/endpoints.py
index 0977b2ab3a..db5ec03df2 100644
--- a/qpid/python/qpid/tests/messaging/endpoints.py
+++ b/qpid/python/qpid/tests/messaging/endpoints.py
@@ -46,6 +46,10 @@ class SetupTests(Base):
     self.conn.open()
     self.ping(self.conn.session())
 
+  def testTcpNodelay(self):
+    self.conn = Connection.establish(self.broker, tcp_nodelay=True)
+    assert self.conn._driver._transport.socket.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)
+
   def testConnectError(self):
     try:
       # Specifying port 0 yields a bad address on Windows; port 4 is unassigned
@@ -111,8 +115,8 @@ class SetupTests(Base):
 
     class flaky:
 
-      def __init__(self, host, port):
-        self.real = real(host, port)
+      def __init__(self, conn, host, port):
+        self.real = real(conn, host, port)
         self.sent_count = 0
         self.recv_count = 0
 
@@ -251,8 +255,8 @@ class ConnectionTests(Base):
 
 class hangable:
 
-  def __init__(self, host, port):
-    self.tcp = TRANSPORTS["tcp"](host, port)
+  def __init__(self, conn, host, port):
+    self.tcp = TRANSPORTS["tcp"](conn, host, port)
     self.hung = False
 
   def hang(self):
@@ -1182,6 +1186,16 @@ test-link-bindings-queue; {
       snd.send(m)
     self.drain(qrcv, expected=msgs)
 
+  def testAssert1(self):
+    try:
+      snd = self.ssn.sender("amq.topic; {assert: always, node: {type: queue}}")
+      assert 0, "assertion failed to trigger"
+    except AssertionFailed, e:
+      pass
+
+  def testAssert2(self):
+    snd = self.ssn.sender("amq.topic; {assert: always}")
+
 NOSUCH_Q = "this-queue-should-not-exist"
 UNPARSEABLE_ADDR = "name/subject; {bad options"
 UNLEXABLE_ADDR = "\0x0\0x1\0x2\0x3"
diff --git a/qpid/tools/src/py/qpid-route b/qpid/tools/src/py/qpid-route
index 516c02d5cd..d98cefd618 100755
--- a/qpid/tools/src/py/qpid-route
+++ b/qpid/tools/src/py/qpid-route
@@ -580,6 +580,9 @@ def main(argv=None):
             else:
                 Usage()
                 return(-1)
+        else:
+            Usage()
+            return(-1)
 
     except Exception,e:
         if rm: