* src/qpid/cluster/SessionFrame.cpp, .h: Wrap AMQFrame with

	  session UUID and direction.
	  
	* src/qpid/cluster/Cluster.cpp, .h: Use SessionFrame.

	* src/qpid/framing/AMQFrame.h, .cpp: Added setBody(), inline getBody()

	* src/qpid/framing/Uuid.h, .cpp:  Clean up constructors, inline.

	* src/qpid/framing/Buffer.h: Put/get byte*, size_T.

	* src/qpid/cluster/SessionManager.cpp, .h:
	 - Maintain the session map.
	 - Handle frames from cluster, dispatch to proper channels.
	 - Implement HandlerUpdater for new channels and maintains


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

4 files changed, 36 insertions, 26 deletions

diff --git a/qpid/cpp/src/tests/Cluster.cpp b/qpid/cpp/src/tests/Cluster.cpp
index ccc1b52249..2e6d9ecfff 100644
--- a/qpid/cpp/src/tests/Cluster.cpp
+++ b/qpid/cpp/src/tests/Cluster.cpp
@@ -36,10 +36,13 @@ using namespace qpid::log;
 BOOST_AUTO_TEST_CASE(testClusterOne) {
     TestCluster cluster("clusterOne", "amqp:one:1");
     AMQFrame frame(VER, 1, new ChannelPingBody(VER));
-    cluster.getToChains().in->handle(frame);
-    BOOST_REQUIRE(cluster.in.waitFor(1));
+    cluster.getSendChains().in->handle(frame);
+    BOOST_REQUIRE(cluster.received.waitFor(1));
 
-    BOOST_CHECK_TYPEID_EQUAL(ChannelPingBody, *cluster.in[0].getBody());
+    SessionFrame& sf=cluster.received[0];
+    BOOST_CHECK(sf.isIncoming);
+    BOOST_CHECK_TYPEID_EQUAL(ChannelPingBody, *sf.frame.getBody());
+    
     BOOST_CHECK_EQUAL(1u, cluster.size());
     Cluster::MemberList members = cluster.getMembers();
     BOOST_CHECK_EQUAL(1u, members.size());
@@ -57,11 +60,13 @@ BOOST_AUTO_TEST_CASE(testClusterTwo) {
 
         // Exchange frames with child.
         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());
+        cluster.getSendChains().in->handle(frame);
+        BOOST_REQUIRE(cluster.received.waitFor(1));
+        SessionFrame& sf=cluster.received[0];
+        BOOST_CHECK(sf.isIncoming);
+        BOOST_CHECK_TYPEID_EQUAL(ChannelPingBody, *sf.frame.getBody());
+        BOOST_REQUIRE(cluster.received.waitFor(2));
+        BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *cluster.received[1].frame.getBody());
 
         // Wait for child to exit.
         int status;
@@ -99,3 +104,4 @@ BOOST_AUTO_TEST_CASE(testClassifierHandlerWiring) {
     BOOST_CHECK_EQUAL(1u, other->count);
 }
     
+
diff --git a/qpid/cpp/src/tests/Cluster.h b/qpid/cpp/src/tests/Cluster.h
index f37c87a9ad..8fddd1d1f7 100644
--- a/qpid/cpp/src/tests/Cluster.h
+++ b/qpid/cpp/src/tests/Cluster.h
@@ -44,10 +44,10 @@ using namespace boost;
 
 void null_deleter(void*) {}
 
-struct TestFrameHandler :
-    public FrameHandler, public vector<AMQFrame>, public Monitor
+template <class T>
+struct TestHandler : public Handler<T&>, public vector<T>, public Monitor
 {
-    void handle(AMQFrame& frame) {
+    void handle(T& frame) {
         Mutex::ScopedLock l(*this);
         push_back(frame);
         notifyAll();
@@ -56,23 +56,22 @@ struct TestFrameHandler :
     bool waitFor(size_t n) {
         Mutex::ScopedLock l(*this);
         AbsTime deadline(now(), 5*TIME_SEC);
-        while (size() != n && wait(deadline))
+        while (vector<T>::size() != n && wait(deadline))
             ;
-        return size() == n;
+        return vector<T>::size() == n;
     }
 };
 
+typedef TestHandler<AMQFrame> TestFrameHandler;
+typedef TestHandler<SessionFrame> TestSessionFrameHandler;
+
 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)
-            ));
+        setReceivedChain(make_shared_ptr(&received, nullDeleter));
     }
 
     /** Wait for cluster to be of size n. */
@@ -80,7 +79,7 @@ struct TestCluster : public Cluster
         return wait(boost::bind(equal_to<size_t>(), bind(&Cluster::size,this), n));
     }
 
-    TestFrameHandler in, out;
+    TestSessionFrameHandler received;
 };
 
 
diff --git a/qpid/cpp/src/tests/Cluster_child.cpp b/qpid/cpp/src/tests/Cluster_child.cpp
index d73d2bdbc7..216afc7bca 100644
--- a/qpid/cpp/src/tests/Cluster_child.cpp
+++ b/qpid/cpp/src/tests/Cluster_child.cpp
@@ -33,13 +33,16 @@ static const ProtocolVersion VER;
 /** Chlid part of Cluster::clusterTwo test */
 void clusterTwo() {
     TestCluster cluster("clusterTwo", "amqp::2");
-    BOOST_REQUIRE(cluster.in.waitFor(1)); // Frame from parent.
-    BOOST_CHECK_TYPEID_EQUAL(ChannelPingBody, *cluster.in[0].getBody());
+    BOOST_REQUIRE(cluster.received.waitFor(1)); // Frame from parent.
+    BOOST_CHECK(cluster.received[0].isIncoming);
+    BOOST_CHECK_TYPEID_EQUAL(ChannelPingBody, *cluster.received[0].frame.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());
+    cluster.getSendChains().out->handle(frame);
+    BOOST_REQUIRE(cluster.received.waitFor(2));
+    BOOST_CHECK(!cluster.received[1].isIncoming);
+    BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *cluster.received[1].frame.getBody());
 } 
 
 int test_main(int, char**) {
diff --git a/qpid/cpp/src/tests/Uuid.cpp b/qpid/cpp/src/tests/Uuid.cpp
index 43d1cbcbba..da8c94aeae 100644
--- a/qpid/cpp/src/tests/Uuid.cpp
+++ b/qpid/cpp/src/tests/Uuid.cpp
@@ -37,6 +37,7 @@ struct UniqueSet : public std::set<Uuid> {
 BOOST_AUTO_TEST_CASE(testUuidCtor) {
     // Uniqueness
     boost::array<Uuid,1000> uuids;
+    for_each(uuids.begin(), uuids.end(), mem_fun_ref(&Uuid::generate));
     UniqueSet unique;
     for_each(uuids.begin(), uuids.end(), unique);
 }
@@ -62,10 +63,11 @@ BOOST_AUTO_TEST_CASE(testUuidOstream) {
 
 BOOST_AUTO_TEST_CASE(testUuidEncodeDecode) {
     Buffer buf(Uuid::size());
-    Uuid uuid;
+    Uuid uuid(sample.c_array());
     uuid.encode(buf);
     buf.flip();
     Uuid decoded;
     decoded.decode(buf);
-    BOOST_CHECK(uuid==decoded);
+    BOOST_CHECK_EQUAL(string(sample.begin(), sample.end()),
+                      string(decoded.begin(), decoded.end()));
 }