Cluster code fixed for changes in codebase.

 - Using SessionManager::Observer 
 - Better ais test setup, only need to be member of ais group.
 - Update cluster_client
 - SessionState holds handler chains.
 - Cluster frames include next handler ptr.


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

10 files changed, 90 insertions, 330 deletions

diff --git a/cpp/src/tests/Cluster.cpp b/cpp/src/tests/Cluster.cpp
deleted file mode 100644
index b448128620..0000000000
--- a/cpp/src/tests/Cluster.cpp
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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 "Cluster.h"
-#include "test_tools.h"
-
-#include "qpid/framing/SessionOpenBody.h"
-#include "qpid/framing/SessionAttachedBody.h"
-#include "qpid/framing/all_method_bodies.h"
-#include "qpid/cluster/ClassifierHandler.h"
-
-#include "unit_test.h"
-
-#include <sys/wait.h>
-
-QPID_AUTO_TEST_SUITE(ClusterTestSuite)
-
-static const ProtocolVersion VER;
-
-/** Verify membership in a cluster with one member. */
-BOOST_AUTO_TEST_CASE(testClusterOne) {
-    TestCluster cluster("clusterOne", "amqp:one:1");
-    AMQFrame send(in_place<SessionOpenBody>(VER));
-    send.setChannel(1);
-    cluster.handle(send);
-    AMQFrame received = cluster.received.pop();
-    BOOST_CHECK_TYPEID_EQUAL(SessionOpenBody, *received.getBody());
-    BOOST_CHECK_EQUAL(1u, cluster.size());
-    Cluster::MemberList members = cluster.getMembers();
-    BOOST_CHECK_EQUAL(1u, members.size());
-    Cluster::Member me=members.front();
-    BOOST_REQUIRE_EQUAL(me.url, "amqp:one:1");
-}
-
-/** Fork a process to test a cluster with two members */
-BOOST_AUTO_TEST_CASE(testClusterTwo) {
-    bool nofork=getenv("NOFORK") != 0;
-    pid_t pid=0;
-    if (!nofork) {
-        pid = fork();
-        BOOST_REQUIRE(pid >= 0);
-    }
-    if (pid || nofork) {        // Parent
-        BOOST_MESSAGE("Parent start");
-        TestCluster cluster("clusterTwo", "amqp:parent:1");
-        BOOST_REQUIRE(cluster.waitFor(2)); // Myself and child.
-
-        // Exchange frames with child.
-        AMQFrame send(SessionOpenBody(VER));
-        send.setChannel(1);
-        cluster.handle(send);
-        AMQFrame received = cluster.received.pop();
-        BOOST_CHECK_TYPEID_EQUAL(SessionOpenBody, *received.getBody());
-        
-        received=cluster.received.pop();
-        BOOST_CHECK_TYPEID_EQUAL(SessionAttachedBody, *received.getBody());
-
-        if (!nofork) {
-            // 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_EQUAL(1u, cluster.size());
-        }
-    }
-    else {                      // Child
-        BOOST_REQUIRE(execl("./Cluster_child", "./Cluster_child", NULL));
-    }
-}
-
-struct CountHandler : public FrameHandler {
-    CountHandler() : count(0) {}
-    void handle(AMQFrame&) { count++; }
-    size_t count;
-};
-    
-/** Test the ClassifierHandler */
-BOOST_AUTO_TEST_CASE(testClassifierHandlerWiring) {
-    AMQFrame queueDecl(in_place<QueueDeclareBody>(VER));
-    AMQFrame messageTrans(in_place<MessageTransferBody>(VER));
-    CountHandler wiring;
-    CountHandler other;
-    
-    ClassifierHandler classify(wiring, other);
-
-    classify.handle(queueDecl);
-    BOOST_CHECK_EQUAL(1u, wiring.count);
-    BOOST_CHECK_EQUAL(0u, other.count);
-    
-    classify.handle(messageTrans);
-    BOOST_CHECK_EQUAL(1u, wiring.count);
-    BOOST_CHECK_EQUAL(1u, other.count);
-}
-
-QPID_AUTO_TEST_SUITE_END()
diff --git a/cpp/src/tests/Cluster.h b/cpp/src/tests/Cluster.h
deleted file mode 100644
index 6ff5c21fdb..0000000000
--- a/cpp/src/tests/Cluster.h
+++ /dev/null
@@ -1,73 +0,0 @@
-#ifndef CLUSTER_H
-#define CLUSTER_H
-
-/*
- *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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/cluster/Cluster.h"
-#include "qpid/sys/BlockingQueue.h"
-#include "qpid/framing/AMQFrame.h"
-
-#include <boost/bind.hpp>
-#include <boost/test/test_tools.hpp>
-
-#include <iostream>
-#include <functional>
-
-/**
- * Definitions for the Cluster.cpp and Cluster_child.cpp child program.
- */
-
-// using namespace in header file is bad manners, but this is strictly for
-// the tests.
-using namespace std;
-using namespace qpid;
-using namespace qpid::cluster;
-using namespace qpid::framing;
-using namespace qpid::sys;
-using namespace boost;
-
-void null_deleter(void*) {}
-
-template <class T>
-class TestHandler : public Handler<T&>, public BlockingQueue<T>
-{
-  public:
-    void handle(T& frame) { push(frame); }
-};
-
-typedef TestHandler<AMQFrame> TestFrameHandler;
-
-struct TestCluster : public Cluster
-{
-    TestCluster(string name, string url)
-        : Cluster(name, url, *(qpid::broker::Broker*)0) {}
-
-    /** Wait for cluster to be of size n. */
-    bool waitFor(size_t n) {
-        BOOST_CHECKPOINT("About to call Cluster::wait");
-        return wait(boost::bind(
-                        equal_to<size_t>(), bind(&Cluster::size,this), n));
-    }
-
-    TestFrameHandler received;
-};
-
-
-
-#endif  /*!CLUSTER_H*/
diff --git a/cpp/src/tests/Cluster_child.cpp b/cpp/src/tests/Cluster_child.cpp
deleted file mode 100644
index 75591bb68f..0000000000
--- a/cpp/src/tests/Cluster_child.cpp
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
- *
- */
-
-// Child process for the Cluster test suite multi-process tests.
-
-#include "Cluster.h"
-#include "test_tools.h"
-#include "qpid/framing/SessionOpenBody.h"
-#include "qpid/framing/SessionAttachedBody.h"
-
-using namespace std;
-using namespace qpid;
-using namespace qpid::cluster;
-using namespace qpid::framing;
-using namespace qpid::sys;
-using namespace qpid::log;
-
-static const ProtocolVersion VER;
-
-/** Child part of Cluster::clusterTwo test */
-void clusterTwo() {
-    TestCluster cluster("clusterTwo", "amqp:child:2");
-    AMQFrame frame = cluster.received.pop(frame); // Frame from parent.
-    BOOST_CHECK_TYPEID_EQUAL(SessionOpenBody, *frame.getBody());
-    BOOST_CHECK_EQUAL(2u, cluster.size()); // Me and parent
-
-    AMQFrame send(in_place<SessionAttachedBody>(VER));
-    send.setChannel(1);
-    cluster.handle(send);
-    BOOST_REQUIRE(cluster.received.waitPop(frame));
-    BOOST_CHECK_TYPEID_EQUAL(SessionAttachedBody, *frame.getBody());
-} 
-
-int test_main(int, char**) {
-    clusterTwo();
-    return 0;
-}
-
diff --git a/cpp/src/tests/Cpg.cpp b/cpp/src/tests/Cpg.cpp
index fddfe9ef05..e8339bf773 100644
--- a/cpp/src/tests/Cpg.cpp
+++ b/cpp/src/tests/Cpg.cpp
@@ -78,12 +78,16 @@ struct Callback : public Cpg::Handler {
         cpg_handle_t /*handle*/,
         struct cpg_name *grp,
         struct cpg_address */*members*/, int nMembers,
-        struct cpg_address */*left*/, int /*nLeft*/,
-        struct cpg_address */*joined*/, int /*nJoined*/
+        struct cpg_address */*left*/, int nLeft,
+        struct cpg_address */*joined*/, int nJoined
     )
     {
         BOOST_CHECK_EQUAL(group, Cpg::str(*grp));
         configChanges.push_back(nMembers);
+        BOOST_MESSAGE("configChange: "<<
+                      nLeft<<" left "<<
+                      nJoined<<" joined "<<
+                      nMembers<<" members.");
     }
 };
 
diff --git a/cpp/src/tests/ais_check b/cpp/src/tests/ais_check
index ce3bbe1b1c..ae0edf88c1 100755
--- a/cpp/src/tests/ais_check
+++ b/cpp/src/tests/ais_check
@@ -2,10 +2,12 @@
 # Check for requirements, run AIS tests if found.
 #
 
-test `id -ng` = "ais" || BADGROUP=yes
-{ ps -u root | grep aisexec >/dev/null; } || NOAISEXEC=yes
+id -nG | grep '\<ais\>' || \
+    NOGROUP="You are not a member of the ais group."
+ps -u root | grep aisexec >/dev/null || \
+    NOAISEXEC="The aisexec daemon is not running as root"
 
-if test -n "$BADGROUP" -o -n "$NOAISEXEC"; then
+if test -n "$NOGROUP" -o -n "$NOAISEXEC"; then
     cat <<EOF
 
     =========== WARNING: NOT RUNNING AIS TESTS ==============
@@ -13,18 +15,8 @@ if test -n "$BADGROUP" -o -n "$NOAISEXEC"; then
     Tests that depend on the openais library (used for clustering)
     will not be run because:
 
-EOF
-    test -n "$BADGROUP" && cat <<EOF
-    You do not appear to have you group ID set to "ais". Make ais your
-    primary group, or run "newgrp ais" before running the tests.
-
-EOF
-    test -n "$NOAISEXEC" && cat <<EOF    
-    The aisexec daemon is not running. Make sure /etc/ais/openais.conf
-    is a valid configuration and aisexec is run by root.
-EOF
-
-    cat <<EOF
+    $NOGROUP
+    $NOAISEXEC
 
     ==========================================================
     
@@ -32,8 +24,4 @@ EOF
     exit 0;			# A warning, not a failure.
 fi
 
-FAILED=0
-for test in `cat ais_tests`; do
-    ./$test || FAILED=`expr $FAILED + 1`
-done
-exit $FAILED
+echo ./ais_run | newgrp ais
diff --git a/cpp/src/tests/ais_run b/cpp/src/tests/ais_run
new file mode 100755
index 0000000000..0f45edc39c
--- /dev/null
+++ b/cpp/src/tests/ais_run
@@ -0,0 +1,15 @@
+#!/bin/sh
+#
+# Run AIS tests, assumes that ais_check has passed and we are
+# running with the ais group ID.
+#
+
+# FIXME aconway 2008-01-30: we should valgrind the cluster brokers.
+
+srcdir=`dirname $0`
+$srcdir/start_cluster 4
+./ais_test
+ret=$?
+$srcdir/stop_cluster 
+exit $ret
+
diff --git a/cpp/src/tests/ais_test.cpp b/cpp/src/tests/ais_test.cpp
new file mode 100644
index 0000000000..00c61242e4
--- /dev/null
+++ b/cpp/src/tests/ais_test.cpp
@@ -0,0 +1,23 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.
+ *
+ */
+
+// Defines test_main function to link with actual unit test code.
+#define BOOST_AUTO_TEST_MAIN	// Boost 1.33
+#define BOOST_TEST_MAIN
+#include "unit_test.h"
+
diff --git a/cpp/src/tests/cluster.mk b/cpp/src/tests/cluster.mk
index a6f7fa90b4..3db0604e2c 100644
--- a/cpp/src/tests/cluster.mk
+++ b/cpp/src/tests/cluster.mk
@@ -1,55 +1,20 @@
-# FIXME aconway 2007-08-31: Disabled cluster compilation,
-# has not been kept up to date with recent commits.
+if CLUSTER
+#
+# Cluster tests makefile fragment, to be included in Makefile.am
 # 
 
-# if CLUSTER
-# # Cluster tests makefile fragment, to be included in Makefile.am
-# # 
+lib_cluster = $(abs_builddir)/../libqpidcluster.la
 
-# lib_cluster = $(abs_builddir)/../libqpidcluster.la
-
-# # NOTE: Programs using the openais library must be run with gid=ais
-# # You should do "newgrp ais" before running the tests to run these.
-# # 
-
-# #
-# # Cluster tests.
-# # 
-
-# # ais_check runs ais if the conditions to run AIS tests
-# # are met, otherwise it prints a warning.
-# TESTS+=ais_check
-# EXTRA_DIST+=ais_check
-# AIS_TESTS=
-
-# ais_check: ais_tests
-# ais_tests:
-# 	echo $(AIS_TESTS)
-# 	echo "# AIS tests" >$@
-# 	for t in $(AIS_TESTS); do echo ./$$t >$@; done
-# 	chmod a+x $@
-
-# CLEANFILES+=ais_tests
-
-# AIS_TESTS+=Cpg
-# check_PROGRAMS+=Cpg
-# Cpg_SOURCES=Cpg.cpp
-# Cpg_LDADD=$(lib_cluster) -lboost_unit_test_framework
-
-# # TODO aconway 2007-07-26: Fix this test.
-# #AIS_TESTS+=Cluster
-# # check_PROGRAMS+=Cluster
-# # Cluster_SOURCES=Cluster.cpp Cluster.h
-# # Cluster_LDADD=$(lib_cluster) -lboost_unit_test_framework
+# NOTE: Programs using the openais library must be run with gid=ais
+# You should do "newgrp ais" before running the tests to run these.
+# 
 
-# check_PROGRAMS+=Cluster_child 
-# Cluster_child_SOURCES=Cluster_child.cpp Cluster.h
-# Cluster_child_LDADD=$(lib_cluster) -lboost_test_exec_monitor
+# ais_check checks conditions for AIS tests and runs if ok.
+TESTS+=ais_check
+EXTRA_DIST+=ais_check ais_run
 
-# # TODO aconway 2007-07-03: In progress
-# #AIS_TESTS+=cluster_client
-# check_PROGRAMS+=cluster_client
-# cluster_client_SOURCES=cluster_client.cpp
-# cluster_client_LDADD=$(lib_client) -lboost_unit_test_framework
+check_PROGRAMS+=ais_test
+ais_test_SOURCES=ais_test.cpp Cpg.cpp 
+ais_test_LDADD=$(lib_client) $(lib_cluster) -lboost_unit_test_framework
 
-# endif
+endif
diff --git a/cpp/src/tests/cluster_client.cpp b/cpp/src/tests/cluster_client.cpp
index c74d7329f0..30b7e38801 100644
--- a/cpp/src/tests/cluster_client.cpp
+++ b/cpp/src/tests/cluster_client.cpp
@@ -16,21 +16,25 @@
  *
  */
 
-#include "qpid/client/Connection.h"
-#include "qpid/shared_ptr.h"
-
 #include "unit_test.h"
+#include "BrokerFixture.h"
+#include "qpid/client/Session.h"
 
 #include <fstream>
 #include <vector>
 #include <functional>
 
-
 QPID_AUTO_TEST_SUITE(cluster_clientTestSuite)
 
-using namespace std;
 using namespace qpid;
 using namespace qpid::client;
+using namespace qpid::framing;
+using namespace qpid::client::arg;
+using framing::TransferContent;
+using std::vector;
+using std::string;
+using std::ifstream;
+using std::ws;
 
 struct ClusterConnections : public vector<shared_ptr<Connection> > {
     ClusterConnections() {
@@ -58,25 +62,23 @@ BOOST_AUTO_TEST_CASE(testWiringReplication) {
     ClusterConnections cluster;
     BOOST_REQUIRE(cluster.size() > 1);
 
-    Exchange fooEx("FooEx", Exchange::TOPIC_EXCHANGE);
-    Queue fooQ("FooQ");
-    
-    Channel broker0;
-    cluster[0]->openChannel(broker0);
-    broker0.declareExchange(fooEx);
-    broker0.declareQueue(fooQ);
-    broker0.bind(fooEx, fooQ, "FooKey");
+    Session broker0 = cluster[0]->newSession();
+    broker0.exchangeDeclare(exchange="ex");
+    broker0.queueDeclare(queue="q");
+    broker0.queueBind(exchange="ex", queue="q", routingKey="key");
     broker0.close();
     
     for (size_t i = 1; i < cluster.size(); ++i) {
-        Channel ch;
-        cluster[i]->openChannel(ch);
-        ch.publish(Message("hello"), fooEx, "FooKey");
-        Message m;
-        BOOST_REQUIRE(ch.get(m, fooQ));
-        BOOST_REQUIRE_EQUAL(m.getData(), "hello");
-        ch.close();
-    }
+        Session s = cluster[i]->newSession();
+        s.messageTransfer(content=TransferContent("data", "key", "ex"));
+        s.messageSubscribe(queue="q", destination="q");
+        s.messageFlow(destination="q", unit=0, value=1);//messages
+        FrameSet::shared_ptr msg = s.get();
+        BOOST_CHECK(msg->isA<MessageTransferBody>());
+        BOOST_CHECK_EQUAL(string("data"), msg->getContent());
+        s.getExecution().completed(msg->getId(), true, true);
+        cluster[i]->close();
+    }    
 }
 
 QPID_AUTO_TEST_SUITE_END()
diff --git a/cpp/src/tests/start_cluster b/cpp/src/tests/start_cluster
index 5992a75d8d..03fb671bdf 100755
--- a/cpp/src/tests/start_cluster
+++ b/cpp/src/tests/start_cluster
@@ -12,7 +12,7 @@ shift
 OPTS=$*
 CLUSTER=`whoami`		# Cluster name=user name, avoid clashes.
 for (( i=0; i<SIZE; ++i )); do
-    PORT=`../qpidd -dp0 --log-output=cluster$i.log --cluster $CLUSTER $OPTS` || exit 1
+    PORT=`../qpidd --load-module ../.libs/libqpidcluster.so -dp0 --log-output=cluster$i.log --cluster-name $CLUSTER $OPTS` || exit 1
     echo $PORT >> cluster.ports
 done