* Summary:

	 - Improved plugin framework and HandlerUpdater interface.
	 - Cluster handlers for traffic to/from cluster.
	 - Cluster HandlerUpdater configures channel chains for cluster.
	 - Cluster PluginProvider registers cluster objects with broker.

	* src/qpid/framing/AMQFrame.h: Made data members public. Handlers
	need to be able to modify frame data, getters/setters are just a
	nuisance here.

	* src/tests/Cluster.cpp: Updated for cluster changes, using
	handlers instead of friendship to hook test into Cluster code.

	* src/qpid/framing/amqp_types.h: Added CHANNEL_MAX and
	CHANNEL_HIGH_BIT constants.

	* src/qpid/framing/HandlerUpdater.h: Renamed ChannelInitializer,
	broke dependency on broker channel types.

	* src/qpid/framing/Handler.h: Added constructors and nextHandler()

	* src/qpid/framing/AMQFrame.h (class AMQFrame): Inlined getChannel()

	* src/qpid/cluster/ClusterPluginProvider.cpp: Provider for cluster
	plugins.

	* src/qpid/cluster/Cluster.cpp: Use ChannelManager. Factor out
	plugin details to ClusterPluginProvider.

	* src/qpid/cluster/ChannelManager.h: Insert cluster handlers
	into channel chains, route frames between cluster and channels.
	
	* src/qpid/broker/BrokerAdapter.cpp (startOk): use CHANNEL_MAX
	constant.

	* src/qpid/broker/Broker.cpp:
	 - Refactored for new plugin framework.
	 - Added getUrl().

	* src/qpid/Url.h: Added constructor from Address.

	* src/qpid/Plugin.h: Generalized plugin framework, broke
	dependency on Broker interfaces. We may want to use plug-ins for
	clients also at some point.

	* src/tests/run_test: Fix bug when VALGRIND is not set.


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@551981 13f79535-47bb-0310-9956-ffa450edef68

10 files changed, 68 insertions, 65 deletions

diff --git a/qpid/cpp/src/tests/.valgrindrc-default b/qpid/cpp/src/tests/.valgrindrc-default
index 41e484d04e..4aba7661de 100644
--- a/qpid/cpp/src/tests/.valgrindrc-default
+++ b/qpid/cpp/src/tests/.valgrindrc-default
@@ -3,5 +3,5 @@
 --demangle=yes
 --suppressions=.valgrind.supp
 --num-callers=25
---track-fds=yes
+--trace-children=yes
 
diff --git a/qpid/cpp/src/tests/BrokerChannelTest.cpp b/qpid/cpp/src/tests/BrokerChannelTest.cpp
index 9dee1fc862..29ed1ae230 100644
--- a/qpid/cpp/src/tests/BrokerChannelTest.cpp
+++ b/qpid/cpp/src/tests/BrokerChannelTest.cpp
@@ -60,7 +60,7 @@ class BrokerChannelTest : public CppUnit::TestCase
     CPPUNIT_TEST(testFlow);
     CPPUNIT_TEST_SUITE_END();
 
-    Broker::shared_ptr broker;
+    shared_ptr<Broker> broker;
     Connection connection;
     MockHandler handler;
     
diff --git a/qpid/cpp/src/tests/Cluster.cpp b/qpid/cpp/src/tests/Cluster.cpp
index ed50cc5d7b..008575140b 100644
--- a/qpid/cpp/src/tests/Cluster.cpp
+++ b/qpid/cpp/src/tests/Cluster.cpp
@@ -24,46 +24,44 @@
 #include "qpid/framing/BasicGetOkBody.h"
 
 
-
 static const ProtocolVersion VER;
 
 /** Verify membership ind a cluster with one member. */
 BOOST_AUTO_TEST_CASE(clusterOne) {
-    VectorFrameHandler received;
-    Cluster cluster("Test", "amqp:one:1", received, VER);
+    Cluster cluster("Test", "amqp:one:1");
+    TestClusterHandler handler(cluster);
     AMQFrame frame(VER, 1, new ChannelOkBody(VER));
-
     cluster.handle(frame);
-    BOOST_REQUIRE(received.waitFor(1));
+    BOOST_REQUIRE(handler.waitFrames(1));
     BOOST_CHECK_EQUAL(1u, cluster.size());
     Cluster::MemberList members = cluster.getMembers();
     BOOST_CHECK_EQUAL(1u, members.size());
     BOOST_REQUIRE_EQUAL(members.front()->url, "amqp:one:1");
-    BOOST_CHECK_EQUAL(1u, received.size());
-    BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *received[0].getBody());
+    BOOST_CHECK_EQUAL(1u, handler.size());
+    BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *handler[0].getBody());
 }
 
 /** Fork a process to verify membership in a cluster with two members */
 BOOST_AUTO_TEST_CASE(clusterTwo) {
-    VectorFrameHandler received;
     pid_t pid=fork();
     BOOST_REQUIRE(pid >= 0);
-    if (pid) {                  // Parent
-        TestCluster cluster("Test", "amqp::1", received, VER);
-        BOOST_REQUIRE(cluster.waitFor(2));
+    if (pid) {                  // Parent see Cluster_child.cpp for child.
+        Cluster cluster("Test", "amqp::1");
+        TestClusterHandler handler(cluster);
+        BOOST_REQUIRE(handler.waitMembers(2));
 
         // Exchange frames with child.
         AMQFrame frame(VER, 1, new ChannelOkBody(VER));
         cluster.handle(frame);
-        BOOST_REQUIRE(received.waitFor(2));
-        BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *received[0].getBody());
-        BOOST_CHECK_TYPEID_EQUAL(BasicGetOkBody, *received[1].getBody());
+        BOOST_REQUIRE(handler.waitFrames(2));
+        BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *handler[0].getBody());
+        BOOST_CHECK_TYPEID_EQUAL(BasicGetOkBody, *handler[1].getBody());
 
         // Wait for child to exit.
         int status;
         BOOST_CHECK_EQUAL(::wait(&status), pid);
         BOOST_CHECK_EQUAL(0, status);
-        BOOST_CHECK(cluster.waitFor(1));
+        BOOST_CHECK(handler.waitMembers(1));
         BOOST_CHECK_EQUAL(1u, cluster.size());
     }
     else {                      // Child
diff --git a/qpid/cpp/src/tests/Cluster.h b/qpid/cpp/src/tests/Cluster.h
index 7ca5445e10..edb1f1524f 100644
--- a/qpid/cpp/src/tests/Cluster.h
+++ b/qpid/cpp/src/tests/Cluster.h
@@ -24,6 +24,7 @@
 #include "qpid/framing/ChannelOkBody.h"
 #include "qpid/framing/BasicGetOkBody.h"
 #include "qpid/log/Logger.h"
+#include <boost/bind.hpp>
 #include <iostream>
 #include <vector>
 
@@ -39,44 +40,44 @@ using namespace qpid::cluster;
 using namespace qpid::framing;
 using namespace qpid::sys;
 
-struct TestCluster : public Cluster {
-    TestCluster(const std::string& name,
-                const std::string& url,
-                framing::FrameHandler& next,
-                framing::ProtocolVersion ver) : Cluster(name,url,next, ver) {}
+void null_deleter(void*) {}
 
-    /** Wait for the cluster to be of expected size (exactly) */
-    bool waitFor(size_t n) {
-        Mutex::ScopedLock l(lock);
-        AbsTime deadline(now(),2*TIME_SEC);
-        while(size() != n && lock.wait(deadline))
-            ;
-        return size() == n;
-    }
-};
-    
-struct VectorFrameHandler :
+struct TestClusterHandler :
     public std::vector<AMQFrame>, public FrameHandler, public Monitor
     
 {
+    TestClusterHandler(Cluster& c) : cluster(c) {
+        cluster.join(make_shared_ptr(this, &null_deleter));
+        cluster.setCallback(boost::bind(&Monitor::notify, this));
+    }
+    
     void handle(AMQFrame& f) {
         ScopedLock l(*this);
         push_back(f);
         notifyAll();
     }
 
-    /** Wait for vector to reach size n exactly */
-    bool waitFor(size_t n) {
+    /** Wait for the vector to contain n frames. */
+    bool waitFrames(size_t n) {
         ScopedLock l(*this);
-        AbsTime deadline(now(), 1*TIME_SEC);
+        AbsTime deadline(now(), TIME_SEC);
         while (size() != n && wait(deadline))
             ;
         return size() == n;
     }
-};
 
+    /** Wait for the cluster to have n members */
+    bool waitMembers(size_t n) {
+        ScopedLock l(*this);
+        AbsTime deadline(now(), TIME_SEC);
+        while (cluster.size() != n && wait(deadline))
+            ;
+        return cluster.size() == n;
+    }
+
+    Cluster& cluster;
+};
 
-// namespace 
 
 
 
diff --git a/qpid/cpp/src/tests/Cluster_child.cpp b/qpid/cpp/src/tests/Cluster_child.cpp
index 1540717f4a..a5ac3e9669 100644
--- a/qpid/cpp/src/tests/Cluster_child.cpp
+++ b/qpid/cpp/src/tests/Cluster_child.cpp
@@ -32,16 +32,14 @@ static const ProtocolVersion VER;
 
 /** Chlid part of Cluster::clusterTwo test */
 void clusterTwo() {
-    VectorFrameHandler received;
-    TestCluster cluster("Test", "amqp::2", received, VER);
-    BOOST_REQUIRE(cluster.waitFor(2));
-    
-    BOOST_REQUIRE(received.waitFor(1));
-    BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *received[0].getBody());
+    Cluster cluster("Test", "amqp::2");
+    TestClusterHandler handler(cluster);
+    BOOST_REQUIRE(handler.waitFrames(1));
+    BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *handler[0].getBody());
     AMQFrame frame(VER, 1, new BasicGetOkBody(VER));
     cluster.handle(frame);
-    BOOST_REQUIRE(received.waitFor(2));
-    BOOST_CHECK_TYPEID_EQUAL(BasicGetOkBody, *received[1].getBody());
+    BOOST_REQUIRE(handler.waitFrames(2));
+    BOOST_CHECK_TYPEID_EQUAL(BasicGetOkBody, *handler[1].getBody());
 } 
 
 int test_main(int, char**) {
diff --git a/qpid/cpp/src/tests/FramingTest.cpp b/qpid/cpp/src/tests/FramingTest.cpp
index 528919de48..9c60af7866 100644
--- a/qpid/cpp/src/tests/FramingTest.cpp
+++ b/qpid/cpp/src/tests/FramingTest.cpp
@@ -402,8 +402,8 @@ class FramingTest : public CppUnit::TestCase
         broker::InProcessBroker::Conversation::const_iterator i = ibroker.conversation.begin();
         ASSERT_FRAME("BROKER: Frame[channel=0; request(id=1,mark=0): ConnectionStart: versionMajor=0; versionMinor=9; serverProperties={}; mechanisms=PLAIN; locales=en_US]", *i++);
         ASSERT_FRAME("CLIENT: Frame[channel=0; response(id=1,request=1,batch=0): ConnectionStartOk: clientProperties={}; mechanism=PLAIN; response=\000guest\000guest; locale=en_US]", *i++);
-        ASSERT_FRAME("BROKER: Frame[channel=0; request(id=2,mark=1): ConnectionTune: channelMax=100; frameMax=65536; heartbeat=0]", *i++);
-        ASSERT_FRAME("CLIENT: Frame[channel=0; response(id=2,request=2,batch=0): ConnectionTuneOk: channelMax=100; frameMax=65536; heartbeat=0]", *i++);
+        ASSERT_FRAME("BROKER: Frame[channel=0; request(id=2,mark=1): ConnectionTune: channelMax=32767; frameMax=65536; heartbeat=0]", *i++);
+        ASSERT_FRAME("CLIENT: Frame[channel=0; response(id=2,request=2,batch=0): ConnectionTuneOk: channelMax=32767; frameMax=65536; heartbeat=0]", *i++);
         ASSERT_FRAME("CLIENT: Frame[channel=0; request(id=1,mark=0): ConnectionOpen: virtualHost=/; capabilities=; insist=1]", *i++);
         ASSERT_FRAME("BROKER: Frame[channel=0; response(id=1,request=1,batch=0): ConnectionOpenOk: knownHosts=]", *i++);
         ASSERT_FRAME("CLIENT: Frame[channel=1; request(id=1,mark=0): ChannelOpen: outOfBand=]", *i++);
diff --git a/qpid/cpp/src/tests/InProcessBroker.h b/qpid/cpp/src/tests/InProcessBroker.h
index 8628bde431..48ac80d30a 100644
--- a/qpid/cpp/src/tests/InProcessBroker.h
+++ b/qpid/cpp/src/tests/InProcessBroker.h
@@ -113,7 +113,7 @@ class InProcessBroker : public client::Connector {
     };
 
     framing::ProtocolInitiation protocolInit;
-    Broker::shared_ptr  broker;
+    shared_ptr<Broker>  broker;
     OutputToInputHandler brokerOut;
     broker::Connection brokerConnection;
     OutputToInputHandler clientOut;
diff --git a/qpid/cpp/src/tests/Makefile.am b/qpid/cpp/src/tests/Makefile.am
index 004693b582..3303afa0be 100644
--- a/qpid/cpp/src/tests/Makefile.am
+++ b/qpid/cpp/src/tests/Makefile.am
@@ -83,7 +83,7 @@ testprogs =		\
   topic_publisher
 
 
-check_PROGRAMS += $(unit_progs) $(testprogs) interop_runner
+check_PROGRAMS += $(unit_progs) $(testprogs) interop_runner 
 
 TESTS_ENVIRONMENT = VALGRIND=$(VALGRIND) srcdir=$(srcdir) $(srcdir)/run_test
 
diff --git a/qpid/cpp/src/tests/Url.cpp b/qpid/cpp/src/tests/Url.cpp
index 09aabb80b3..faf802ba1e 100644
--- a/qpid/cpp/src/tests/Url.cpp
+++ b/qpid/cpp/src/tests/Url.cpp
@@ -30,27 +30,33 @@ BOOST_AUTO_TEST_CASE(testUrl_str) {
     Url url;
     url.push_back(TcpAddress("foo.com"));
     url.push_back(TcpAddress("bar.com", 6789));
-    
-    BOOST_CHECK_EQUAL(
-        url.str(), "amqp:tcp:foo.com:5672,tcp:bar.com:6789");
-    BOOST_CHECK_EQUAL(Url().str(), "amqp:");
+    BOOST_CHECK_EQUAL("amqp:tcp:foo.com:5672,tcp:bar.com:6789", url.str());
+    BOOST_CHECK_EQUAL("amqp:", Url().str());
 }
 
 
-BOOST_AUTO_TEST_CASE(testUrl_ctor) {
-    BOOST_CHECK_EQUAL(
-        Url("amqp:foo.com,tcp:bar.com:1234").str(),
-        "amqp:tcp:foo.com:5672,tcp:bar.com:1234");
-    BOOST_CHECK_EQUAL(
-        Url("amqp:foo/ignorethis").str(),
-        "amqp:tcp:foo:5672");
-    BOOST_CHECK_EQUAL("amqp:tcp::5672", Url("amqp:").str());
-    BOOST_CHECK_EQUAL(0u, Url("xxx", nothrow).size());
+BOOST_AUTO_TEST_CASE(testUrl_parse) {
+    Url url;
+    url.parse("amqp:foo.com,tcp:bar.com:1234");
+    BOOST_CHECK_EQUAL(2u, url.size());
+    BOOST_CHECK_EQUAL("foo.com", boost::get<TcpAddress>(url[0]).host);
+    BOOST_CHECK_EQUAL("amqp:tcp:foo.com:5672,tcp:bar.com:1234", url.str());
+
+    url.parse("amqp:foo/ignorethis");
+    BOOST_CHECK_EQUAL("amqp:tcp:foo:5672", url.str());
+
+    url.parse("amqp:");
+    BOOST_CHECK_EQUAL("amqp:tcp::5672", url.str());
+
     try {
-        Url invalid("xxx");
+        url.parse("invalid url");
         BOOST_FAIL("Expected InvalidUrl exception");
     }
     catch (const Url::InvalidUrl&) {}
+
+    url.parseNoThrow("invalid url");
+    BOOST_CHECK(url.empty());
 }
 
 
+
diff --git a/qpid/cpp/src/tests/run_test b/qpid/cpp/src/tests/run_test
index deb22b4450..ec724fe727 100755
--- a/qpid/cpp/src/tests/run_test
+++ b/qpid/cpp/src/tests/run_test
@@ -47,7 +47,7 @@ if grep -l "^# Generated by .*libtool" "$1" >/dev/null 2>&1; then
     # This is a libtool "executable". Valgrind it if VALGRIND specified.
     test -n "$VALGRIND" && VALGRIND="$VALGRIND --log-file-exactly=$VG_LOG --"
     # Hide output unless there's an error.
-    libtool --mode=execute "$VALGRIND" "$@" >$TEST_LOG 2>&1 || {
+    libtool --mode=execute $VALGRIND "$@" >$TEST_LOG 2>&1 || {
 	ERROR=$?
 	cat $TEST_LOG  
     }