2007-06-30 <aconway@redhat.com>

	* src/qpid/cluster/Cluster.cpp: Refactor - expose 4 handler points
	for all traffic to/from cluster. Removed HandlerUpdater functionality,
	separate class. Cluster only deals with membership and connecting
	the 4 handler points to CPG multicast.

	* src/tests/cluster.mk: Dropped newgrp ais wrapper scripts, its
	much simpler if the user just does "newgrp ais" before building.

	* src/tests/ais_check: Test script to check if users gid is ais
	and give clear notice if not.

	* src/tests/Cluster.cpp: Updated for changes to Cluster. 

	* src/qpid/cluster/Cpg.cpp: Better messages for common errors.

	* Handler.h: Remove nextHandler() minor convenience is outweighted
	by risk of undetected errors if handlers that expect next() to be
	set are called when it's not set.

	* src/qpid/cluster/Cpg.cpp: Added logging. Replaced boost::function
	with traditional virtual interface (nasty stack traces.)


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

7 files changed, 108 insertions, 99 deletions

diff --git a/qpid/cpp/src/tests/Cluster.cpp b/qpid/cpp/src/tests/Cluster.cpp
index 008575140b..2ec140b924 100644
--- a/qpid/cpp/src/tests/Cluster.cpp
+++ b/qpid/cpp/src/tests/Cluster.cpp
@@ -20,51 +20,52 @@
 #include <boost/test/auto_unit_test.hpp>
 #include "test_tools.h"
 #include "Cluster.h"
+#include "qpid/framing/ChannelPingBody.h"
 #include "qpid/framing/ChannelOkBody.h"
-#include "qpid/framing/BasicGetOkBody.h"
-
 
 static const ProtocolVersion VER;
 
-/** Verify membership ind a cluster with one member. */
+using namespace qpid::log;
+
+/** Verify membership in a cluster with one member. */
 BOOST_AUTO_TEST_CASE(clusterOne) {
-    Cluster cluster("Test", "amqp:one:1");
-    TestClusterHandler handler(cluster);
-    AMQFrame frame(VER, 1, new ChannelOkBody(VER));
-    cluster.handle(frame);
-    BOOST_REQUIRE(handler.waitFrames(1));
+    TestCluster cluster("clusterOne", "amqp:one:1");
+    AMQFrame frame(VER, 1, new ChannelPingBody(VER));
+    cluster.getToChains().in->handle(frame);
+    BOOST_REQUIRE(cluster.in.waitFor(1));
+
+    BOOST_CHECK_TYPEID_EQUAL(ChannelPingBody, *cluster.in[0].getBody());
     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, handler.size());
-    BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *handler[0].getBody());
+    shared_ptr<const Cluster::Member> me=members.front();
+    BOOST_REQUIRE_EQUAL(me->url, "amqp:one:1");
 }
 
-/** Fork a process to verify membership in a cluster with two members */
+/** Fork a process to test a cluster with two members */
 BOOST_AUTO_TEST_CASE(clusterTwo) {
     pid_t pid=fork();
     BOOST_REQUIRE(pid >= 0);
-    if (pid) {                  // Parent see Cluster_child.cpp for child.
-        Cluster cluster("Test", "amqp::1");
-        TestClusterHandler handler(cluster);
-        BOOST_REQUIRE(handler.waitMembers(2));
+    if (pid) {              // Parent, see Cluster_child.cpp for child.
+        TestCluster cluster("clusterTwo", "amqp::1");
+        BOOST_REQUIRE(cluster.waitFor(2)); // Myself and child.
 
         // Exchange frames with child.
-        AMQFrame frame(VER, 1, new ChannelOkBody(VER));
-        cluster.handle(frame);
-        BOOST_REQUIRE(handler.waitFrames(2));
-        BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *handler[0].getBody());
-        BOOST_CHECK_TYPEID_EQUAL(BasicGetOkBody, *handler[1].getBody());
+        AMQFrame frame(VER, 1, new ChannelPingBody(VER));
+        cluster.getToChains().in->handle(frame);
+        BOOST_REQUIRE(cluster.in.waitFor(1));
+        BOOST_CHECK_TYPEID_EQUAL(ChannelPingBody, *cluster.in[0].getBody());
+        BOOST_REQUIRE(cluster.out.waitFor(1));
+        BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *cluster.out[0].getBody());
 
         // Wait for child to exit.
         int status;
         BOOST_CHECK_EQUAL(::wait(&status), pid);
         BOOST_CHECK_EQUAL(0, status);
-        BOOST_CHECK(handler.waitMembers(1));
+        BOOST_CHECK(cluster.waitFor(1));
         BOOST_CHECK_EQUAL(1u, cluster.size());
     }
     else {                      // Child
-        BOOST_REQUIRE(execl("Cluster_child", "Cluster_child", NULL));
+        BOOST_REQUIRE(execl("./Cluster_child", "./Cluster_child", NULL));
     }
 }
diff --git a/qpid/cpp/src/tests/Cluster.h b/qpid/cpp/src/tests/Cluster.h
index edb1f1524f..f37c87a9ad 100644
--- a/qpid/cpp/src/tests/Cluster.h
+++ b/qpid/cpp/src/tests/Cluster.h
@@ -27,6 +27,7 @@
 #include <boost/bind.hpp>
 #include <iostream>
 #include <vector>
+#include <functional>
 
 /**
  * Definitions for the Cluster.cpp and Cluster_child.cpp child program.
@@ -39,45 +40,48 @@ using namespace qpid;
 using namespace qpid::cluster;
 using namespace qpid::framing;
 using namespace qpid::sys;
+using namespace boost;
 
 void null_deleter(void*) {}
 
-struct TestClusterHandler :
-    public std::vector<AMQFrame>, public FrameHandler, public Monitor
-    
+struct TestFrameHandler :
+    public FrameHandler, public vector<AMQFrame>, 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);
+    void handle(AMQFrame& frame) {
+        Mutex::ScopedLock l(*this);
+        push_back(frame);
         notifyAll();
     }
 
-    /** Wait for the vector to contain n frames. */
-    bool waitFrames(size_t n) {
-        ScopedLock l(*this);
-        AbsTime deadline(now(), TIME_SEC);
+    bool waitFor(size_t n) {
+        Mutex::ScopedLock l(*this);
+        AbsTime deadline(now(), 5*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;
+void nullDeleter(void*) {}
+
+struct TestCluster : public Cluster
+{
+    TestCluster(string name, string url) : Cluster(name, url)
+    {
+        setFromChains(
+            FrameHandler::Chains(
+                make_shared_ptr(&in, nullDeleter),
+                make_shared_ptr(&out, nullDeleter)
+            ));
     }
 
-    Cluster& cluster;
-};
+    /** Wait for cluster to be of size n. */
+    bool waitFor(size_t n) {
+        return wait(boost::bind(equal_to<size_t>(), bind(&Cluster::size,this), n));
+    }
 
+    TestFrameHandler in, out;
+};
 
 
 
diff --git a/qpid/cpp/src/tests/Cluster_child.cpp b/qpid/cpp/src/tests/Cluster_child.cpp
index a5ac3e9669..d73d2bdbc7 100644
--- a/qpid/cpp/src/tests/Cluster_child.cpp
+++ b/qpid/cpp/src/tests/Cluster_child.cpp
@@ -26,20 +26,20 @@ using namespace qpid;
 using namespace qpid::cluster;
 using namespace qpid::framing;
 using namespace qpid::sys;
-
+using namespace qpid::log;
 
 static const ProtocolVersion VER;
 
 /** Chlid part of Cluster::clusterTwo test */
 void clusterTwo() {
-    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(handler.waitFrames(2));
-    BOOST_CHECK_TYPEID_EQUAL(BasicGetOkBody, *handler[1].getBody());
+    TestCluster cluster("clusterTwo", "amqp::2");
+    BOOST_REQUIRE(cluster.in.waitFor(1)); // Frame from parent.
+    BOOST_CHECK_TYPEID_EQUAL(ChannelPingBody, *cluster.in[0].getBody());
+    BOOST_CHECK_EQUAL(2u, cluster.size()); // Me and parent
+    AMQFrame frame(VER, 1, new ChannelOkBody(VER));
+    cluster.getToChains().out->handle(frame);
+    BOOST_REQUIRE(cluster.out.waitFor(1));
+    BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *cluster.out[0].getBody());
 } 
 
 int test_main(int, char**) {
diff --git a/qpid/cpp/src/tests/Cpg.cpp b/qpid/cpp/src/tests/Cpg.cpp
index 97b829ea63..ec98ca4fc2 100644
--- a/qpid/cpp/src/tests/Cpg.cpp
+++ b/qpid/cpp/src/tests/Cpg.cpp
@@ -47,11 +47,11 @@ ostream& operator<<(ostream& o, const pair<T*, int>& array) {
     o << "{ ";
     ostream_iterator<cpg_address> i(o, " ");
     copy(array.first, array.first+array.second, i);
-    cout << "}";
+    o << "}";
     return o;
 }
 
-struct Callback {
+struct Callback : public Cpg::Handler {
     Callback(const string group_) : group(group_) {}
     string group;
     vector<string> delivered;
@@ -88,10 +88,7 @@ BOOST_AUTO_TEST_CASE(Cpg_basic) {
     //
     Cpg::Name group("foo");
     Callback cb(group.str());
-    Cpg::DeliverFn deliver=boost::bind(&Callback::deliver, &cb, _1, _2, _3, _4, _5, _6);
-    Cpg::ConfigChangeFn reconfig=boost::bind<void>(&Callback::configChange, &cb, _1, _2, _3, _4, _5, _6, _7, _8);
-
-    Cpg cpg(deliver, reconfig);
+    Cpg cpg(cb);
     cpg.join(group);
     iovec iov = { (void*)"Hello!", 6 };
     cpg.mcast(group, &iov, 1);
diff --git a/qpid/cpp/src/tests/Makefile.am b/qpid/cpp/src/tests/Makefile.am
index 3303afa0be..5a8ad1bde6 100644
--- a/qpid/cpp/src/tests/Makefile.am
+++ b/qpid/cpp/src/tests/Makefile.am
@@ -11,17 +11,19 @@ lib_broker = $(abs_builddir)/../libqpidbroker.la
 # Initialize variables that are incremented with +=
 # 
 check_PROGRAMS=
-unit_progs=
-unit_wrappers=
+TESTS=
+EXTRA_DIST=
 
 #
 # Unit test programs.
 # 
-unit_progs+=logging
+TESTS+=logging
+check_PROGRAMS+=logging
 logging_SOURCES=logging.cpp test_tools.h
 logging_LDADD=-lboost_unit_test_framework -lboost_regex $(lib_common) 
 
-unit_progs+=Url
+TESTS+=Url
+check_PROGRAMS+=Url
 Url_SOURCES=Url.cpp test_tools.h
 Url_LDADD=-lboost_unit_test_framework $(lib_common) 
 
@@ -76,21 +78,20 @@ unit_tests =		\
 
 # Executables for client tests
 
-testprogs =		\
+testprogs=		\
   client_test		\
   echo_service		\
   topic_listener	\
-  topic_publisher
-
-
-check_PROGRAMS += $(unit_progs) $(testprogs) interop_runner 
+  topic_publisher	\
+  interop_runner 
+check_PROGRAMS += $(testprogs)
 
 TESTS_ENVIRONMENT = VALGRIND=$(VALGRIND) srcdir=$(srcdir) $(srcdir)/run_test
 
 system_tests = client_test quick_topictest
-TESTS = dummy_test $(unit_progs) $(unit_wrappers) run-unit-tests start_broker $(system_tests) python_tests kill_broker 
+TESTS += run-unit-tests start_broker $(system_tests) python_tests kill_broker 
 
-EXTRA_DIST =								\
+EXTRA_DIST +=								\
   test_env run_test							\
   run-unit-tests start_broker python_tests kill_broker 			\
   quick_topictest							\
@@ -131,10 +132,8 @@ gen.mk: Makefile.am
 check-unit:
 	$(MAKE) check TESTS=$(UNIT_TESTS) run-unit-tests
 
-# Dummy test to force necessary test files to be generated.
-dummy_test: .valgrind.supp .valgrindrc
-	{ echo "#!/bin/sh";  echo "# Dummy test, does nothing. "; } > $@
-	chmod a+x $@
+# Make sure valgrind files are generated.
+all: .valgrind.supp .valgrindrc
 
 # Create a copy so that can be modified without risk of committing the changes.
 .valgrindrc: .valgrindrc-default
diff --git a/qpid/cpp/src/tests/ais_check b/qpid/cpp/src/tests/ais_check
new file mode 100755
index 0000000000..df40899065
--- /dev/null
+++ b/qpid/cpp/src/tests/ais_check
@@ -0,0 +1,16 @@
+#!/bin/sh
+test `id -ng` = "ais" || {
+    cat <<EOF
+    =========================== NOTICE==============================
+
+    You do not appear to have you group ID set to "ais".
+
+    Cluster tests that require the openais library will fail.Make sure
+    you are a member of group ais and run "newgrp ais" before running
+    the tests.
+
+    ================================================================
+    
+EOF
+exit 1;
+}
diff --git a/qpid/cpp/src/tests/cluster.mk b/qpid/cpp/src/tests/cluster.mk
index 6f59b13107..33e1569d3c 100644
--- a/qpid/cpp/src/tests/cluster.mk
+++ b/qpid/cpp/src/tests/cluster.mk
@@ -5,33 +5,25 @@ if CLUSTER
 lib_cluster = $(abs_builddir)/../libqpidcluster.la
 
 # NOTE: Programs using the openais library must be run with gid=ais
-# Such programs are built as *.ais, with a wrapper script *.sh that
-# runs the program with newgrp ais.
+# You should do "newgrp ais" before running the tests to run these.
 # 
 
-# Rule to generate wrapper scripts for tests that require gid=ais.
-run_test="env VALGRIND=$(VALGRIND) srcdir=$(srcdir) $(srcdir)/run_test"
-.ais.sh:
-	echo "if groups | grep '\bais\b' >/dev/null;" > $@_t
-	echo "then echo $(run_test) ./$< \"$$@	\"| newgrp ais;" >>$@_t
-	echo "else echo WARNING: `whoami` not in group ais, skipping $<.;" >>$@_t
-	echo "fi"  >> $@_t
-	mv $@_t $@
-	chmod a+x $@
-
 #
 # Cluster tests.
 # 
-check_PROGRAMS+=Cpg.ais
-Cpg_ais_SOURCES=Cpg.cpp
-Cpg_ais_LDADD=$(lib_cluster) -lboost_unit_test_framework
-unit_wrappers+=Cpg.sh
-
-# FIXME aconway 2007-06-29: Fixing problems with the test.
-# check_PROGRAMS+=Cluster.ais
-# Cluster_ais_SOURCES=Cluster.cpp Cluster.h
-# Cluster_ais_LDADD=$(lib_cluster) -lboost_unit_test_framework
-# unit_wrappers+=Cluster.sh
+
+TESTS+=ais_check
+EXTRA_DIST+=ais_check
+
+TESTS+=Cpg
+check_PROGRAMS+=Cpg
+Cpg_SOURCES=Cpg.cpp
+Cpg_LDADD=$(lib_cluster) -lboost_unit_test_framework
+
+TESTS+=Cluster
+check_PROGRAMS+=Cluster
+Cluster_SOURCES=Cluster.cpp Cluster.h
+Cluster_LDADD=$(lib_cluster) -lboost_unit_test_framework
 
 check_PROGRAMS+=Cluster_child 
 Cluster_child_SOURCES=Cluster_child.cpp Cluster.h