10 files changed, 789 insertions, 45 deletions

diff --git a/qpid/cpp/src/tests/CMakeLists.txt b/qpid/cpp/src/tests/CMakeLists.txt
index 34f5d35a9a..b1219aad74 100644
--- a/qpid/cpp/src/tests/CMakeLists.txt
+++ b/qpid/cpp/src/tests/CMakeLists.txt
@@ -95,6 +95,7 @@ set(unit_tests_to_build
     InlineAllocator
     InlineVector
     ClientSessionTest
+    MessagingSessionTests
     SequenceSet
     StringUtils
     IncompleteMessageList
@@ -128,6 +129,7 @@ set(unit_tests_to_build
     ReplicationTest
     ClientMessageTest
     PollableCondition
+    Variant
     ${xml_tests}
     CACHE STRING "Which unit tests to build"
    )
diff --git a/qpid/cpp/src/tests/FieldTable.cpp b/qpid/cpp/src/tests/FieldTable.cpp
index a02bbd5194..5b43871f6d 100644
--- a/qpid/cpp/src/tests/FieldTable.cpp
+++ b/qpid/cpp/src/tests/FieldTable.cpp
@@ -22,6 +22,7 @@
 #include "qpid/framing/Array.h"
 #include "qpid/framing/FieldTable.h"
 #include "qpid/framing/FieldValue.h"
+#include "qpid/framing/List.h"
 #include "qpid/sys/alloca.h"
 
 #include "unit_test.h"
@@ -86,7 +87,9 @@ QPID_AUTO_TEST_CASE(testAssignment)
 
 QPID_AUTO_TEST_CASE(testNestedValues)
 {
-    char buff[100];
+    double d = 1.2345;
+    uint32_t u = 101;
+    char buff[1000];
     {
         FieldTable a;
         FieldTable b;
@@ -94,11 +97,17 @@ QPID_AUTO_TEST_CASE(testNestedValues)
         items.push_back("one");
         items.push_back("two");
         Array c(items);
+        List list;
+        list.push_back(List::ValuePtr(new Str16Value("red")));
+        list.push_back(List::ValuePtr(new Unsigned32Value(u)));
+        list.push_back(List::ValuePtr(new Str8Value("yellow")));
+        list.push_back(List::ValuePtr(new DoubleValue(d)));
         
         a.setString("id", "A");
         b.setString("id", "B");
         a.setTable("B", b);
         a.setArray("C", c);
+        a.set("my-list", FieldTable::ValuePtr(new ListValue(list)));
 
 
         Buffer wbuffer(buff, 100);
@@ -119,6 +128,27 @@ QPID_AUTO_TEST_CASE(testNestedValues)
         BOOST_CHECK((uint) 2 == items.size());
         BOOST_CHECK(string("one") == items[0]);
         BOOST_CHECK(string("two") == items[1]);
+
+        List list;
+        BOOST_CHECK(a.get("my-list")->get<List>(list));
+        List::const_iterator i = list.begin();
+        BOOST_CHECK(i != list.end());
+        BOOST_CHECK_EQUAL(std::string("red"), (*i)->get<std::string>());
+
+        i++;
+        BOOST_CHECK(i != list.end());
+        BOOST_CHECK_EQUAL(u, (uint32_t) (*i)->get<int>());
+
+        i++;
+        BOOST_CHECK(i != list.end());
+        BOOST_CHECK_EQUAL(std::string("yellow"), (*i)->get<std::string>());
+
+        i++;
+        BOOST_CHECK(i != list.end());
+        BOOST_CHECK_EQUAL(d, (*i)->get<double>());
+
+        i++;
+        BOOST_CHECK(i == list.end());
     }
 }
 
diff --git a/qpid/cpp/src/tests/Makefile.am b/qpid/cpp/src/tests/Makefile.am
index db4c8ba914..a15ba3578c 100644
--- a/qpid/cpp/src/tests/Makefile.am
+++ b/qpid/cpp/src/tests/Makefile.am
@@ -65,6 +65,7 @@ unit_test_LDADD=-lboost_unit_test_framework -lboost_regex  \
 	$(lib_client) $(lib_broker) $(lib_console)
 
 unit_test_SOURCES= unit_test.cpp unit_test.h \
+	MessagingSessionTests.cpp \
 	ClientSessionTest.cpp \
 	BrokerFixture.h SocketProxy.h \
 	exception_test.cpp \
@@ -111,7 +112,8 @@ unit_test_SOURCES= unit_test.cpp unit_test.h \
 	FrameDecoder.cpp \
 	ReplicationTest.cpp \
 	ClientMessageTest.cpp \
-	PollableCondition.cpp
+	PollableCondition.cpp \
+	Variant.cpp
 
 if HAVE_XML
 unit_test_SOURCES+= XmlClientSessionTest.cpp
@@ -268,6 +270,11 @@ check_PROGRAMS+=qrsh
 qrsh_SOURCES=qrsh.cpp
 qrsh_LDADD=$(lib_client)
 
+check_PROGRAMS+=qpid_stream
+qpid_stream_INCLUDES=$(PUBLIC_INCLUDES)
+qpid_stream_SOURCES=qpid_stream.cpp
+qpid_stream_LDADD=$(lib_client) 
+
 
 TESTS_ENVIRONMENT = \
     VALGRIND=$(VALGRIND) \
diff --git a/qpid/cpp/src/tests/MessagingSessionTests.cpp b/qpid/cpp/src/tests/MessagingSessionTests.cpp
new file mode 100644
index 0000000000..4ee27f0764
--- /dev/null
+++ b/qpid/cpp/src/tests/MessagingSessionTests.cpp
@@ -0,0 +1,360 @@
+/*
+ *
+ * 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 "unit_test.h"
+#include "test_tools.h"
+#include "BrokerFixture.h"
+#include "qpid/messaging/Connection.h"
+#include "qpid/messaging/Message.h"
+#include "qpid/messaging/MessageListener.h"
+#include "qpid/messaging/Receiver.h"
+#include "qpid/messaging/Sender.h"
+#include "qpid/messaging/Session.h"
+#include "qpid/client/Connection.h"
+#include "qpid/client/Session.h"
+#include "qpid/framing/reply_exceptions.h"
+#include "qpid/sys/Time.h"
+#include <boost/assign.hpp>
+#include <boost/format.hpp>
+#include <string>
+#include <vector>
+
+QPID_AUTO_TEST_SUITE(MessagingSessionTests)
+
+using namespace qpid::messaging;
+using namespace qpid;
+using qpid::broker::Broker;
+
+struct BrokerAdmin
+{
+    qpid::client::Connection connection;
+    qpid::client::Session session;
+
+    BrokerAdmin(uint16_t port)
+    {
+        connection.open("localhost", port);
+        session = connection.newSession();
+    }
+
+    void createQueue(const std::string& name)
+    {
+        session.queueDeclare(qpid::client::arg::queue=name);
+    }
+
+    void deleteQueue(const std::string& name)
+    {
+        session.queueDelete(qpid::client::arg::queue=name);
+    }
+
+    void createExchange(const std::string& name, const std::string& type)
+    {
+        session.exchangeDeclare(qpid::client::arg::exchange=name, qpid::client::arg::type=type);
+    }
+
+    void deleteExchange(const std::string& name)
+    {
+        session.exchangeDelete(qpid::client::arg::exchange=name);
+    }
+
+    ~BrokerAdmin()
+    {
+        session.close();
+        connection.close();
+    }
+};
+
+struct MessagingFixture : public BrokerFixture
+{
+    Connection connection;
+    Session session;
+    BrokerAdmin admin;
+
+    MessagingFixture(Broker::Options opts = Broker::Options()) : 
+        BrokerFixture(opts),
+        connection(Connection::open((boost::format("amqp:tcp:localhost:%1%") % (broker->getPort(Broker::TCP_TRANSPORT))).str())),
+        session(connection.newSession()),
+        admin(broker->getPort(Broker::TCP_TRANSPORT)) {}
+
+    ~MessagingFixture()
+    {
+        session.close();
+        connection.close();
+    }
+};
+
+struct QueueFixture : MessagingFixture
+{
+    std::string queue;
+
+    QueueFixture(const std::string& name = "test-queue") : queue(name)
+    {
+        admin.createQueue(queue);
+    }
+
+    ~QueueFixture()
+    {
+        admin.deleteQueue(queue);
+    }
+
+};
+
+struct TopicFixture : MessagingFixture
+{
+    std::string topic;
+
+    TopicFixture(const std::string& name = "test-topic", const std::string& type="fanout") : topic(name)
+    {
+        admin.createExchange(topic, type);
+    }
+
+    ~TopicFixture()
+    {
+        admin.deleteExchange(topic);
+    }
+
+};
+
+struct MultiQueueFixture : MessagingFixture
+{
+    typedef std::vector<std::string>::const_iterator const_iterator; 
+    std::vector<std::string> queues;
+
+    MultiQueueFixture(const std::vector<std::string>& names = boost::assign::list_of<std::string>("q1")("q2")("q3")) : queues(names)
+    {
+        for (const_iterator i = queues.begin(); i != queues.end(); ++i) {
+            admin.createQueue(*i);
+        }
+    }
+
+    ~MultiQueueFixture()
+    {
+        for (const_iterator i = queues.begin(); i != queues.end(); ++i) {
+            admin.deleteQueue(*i);
+        }
+    }
+
+};
+
+struct MessageDataCollector : MessageListener
+{
+    std::vector<std::string> messageData;
+
+    void received(Message& message) {
+        messageData.push_back(message.getBytes());
+    }
+};
+
+std::vector<std::string> fetch(Receiver& receiver, int count, qpid::sys::Duration timeout=qpid::sys::TIME_SEC*5) 
+{
+    std::vector<std::string> data;
+    Message message;
+    for (int i = 0; i < count && receiver.fetch(message, timeout); i++) {
+        data.push_back(message.getBytes());
+    }
+    return data;
+}
+
+QPID_AUTO_TEST_CASE(testSimpleSendReceive)
+{
+    QueueFixture fix;
+    Sender sender = fix.session.createSender(fix.queue);
+    Message out("test-message");
+    sender.send(out);
+    Receiver receiver = fix.session.createReceiver(fix.queue);
+    Message in = receiver.fetch(5 * qpid::sys::TIME_SEC);
+    fix.session.acknowledge();
+    BOOST_CHECK_EQUAL(in.getBytes(), out.getBytes());
+}
+
+QPID_AUTO_TEST_CASE(testSendReceiveHeaders)
+{
+    QueueFixture fix;
+    Sender sender = fix.session.createSender(fix.queue);
+    Message out("test-message");
+    for (uint i = 0; i < 10; ++i) {
+        out.getHeaders()["a"] = i; 
+        sender.send(out);
+    }
+    Receiver receiver = fix.session.createReceiver(fix.queue);
+    Message in;
+    for (uint i = 0; i < 10; ++i) {
+        //Message in = receiver.fetch(5 * qpid::sys::TIME_SEC);
+        BOOST_CHECK(receiver.fetch(in, 5 * qpid::sys::TIME_SEC));
+        BOOST_CHECK_EQUAL(in.getBytes(), out.getBytes());
+        BOOST_CHECK_EQUAL(in.getHeaders()["a"].asUint32(), i);
+        fix.session.acknowledge();
+    }
+}
+
+QPID_AUTO_TEST_CASE(testSenderError)
+{
+    MessagingFixture fix;
+    //TODO: this is the wrong type for the exception; define explicit set in messaging namespace
+    BOOST_CHECK_THROW(fix.session.createSender("NonExistentAddress"), qpid::framing::NotFoundException);
+}
+
+QPID_AUTO_TEST_CASE(testReceiverError)
+{
+    MessagingFixture fix;
+    //TODO: this is the wrong type for the exception; define explicit set in messaging namespace
+    BOOST_CHECK_THROW(fix.session.createReceiver("NonExistentAddress"), qpid::framing::NotFoundException);
+}
+
+QPID_AUTO_TEST_CASE(testSimpleTopic)
+{
+    TopicFixture fix;
+
+    Sender sender = fix.session.createSender(fix.topic);
+    Message msg("one");
+    sender.send(msg);
+    Receiver sub1 = fix.session.createReceiver(fix.topic);
+    sub1.setCapacity(10u);
+    sub1.start();
+    msg.setBytes("two");
+    sender.send(msg);
+    Receiver sub2 = fix.session.createReceiver(fix.topic);
+    sub2.setCapacity(10u);
+    sub2.start();
+    msg.setBytes("three");
+    sender.send(msg);
+    Receiver sub3 = fix.session.createReceiver(fix.topic);
+    sub3.setCapacity(10u);
+    sub3.start();
+    msg.setBytes("four");
+    sender.send(msg);
+    BOOST_CHECK_EQUAL(fetch(sub2, 2), boost::assign::list_of<std::string>("three")("four"));
+    sub2.cancel();
+
+    msg.setBytes("five");
+    sender.send(msg);
+    BOOST_CHECK_EQUAL(fetch(sub1, 4), boost::assign::list_of<std::string>("two")("three")("four")("five"));
+    BOOST_CHECK_EQUAL(fetch(sub3, 2), boost::assign::list_of<std::string>("four")("five"));
+    Message in;
+    BOOST_CHECK(!sub2.fetch(in, 0));//TODO: or should this raise an error?
+
+    
+    //TODO: check pending messages...
+}
+
+QPID_AUTO_TEST_CASE(testSessionFetch)
+{
+    MultiQueueFixture fix;
+    
+    for (uint i = 0; i < fix.queues.size(); i++) {
+        Receiver r = fix.session.createReceiver(fix.queues[i]);
+        r.setCapacity(10u);
+        r.start();//TODO: add Session::start
+    }
+
+    for (uint i = 0; i < fix.queues.size(); i++) {
+        Sender s = fix.session.createSender(fix.queues[i]);
+        Message msg((boost::format("Message_%1%") % (i+1)).str());
+        s.send(msg);
+    }    
+    
+    for (uint i = 0; i < fix.queues.size(); i++) {
+        Message msg;
+        BOOST_CHECK(fix.session.fetch(msg, qpid::sys::TIME_SEC));
+        BOOST_CHECK_EQUAL(msg.getBytes(), (boost::format("Message_%1%") % (i+1)).str());
+    }
+}
+
+QPID_AUTO_TEST_CASE(testSessionDispatch)
+{
+    MultiQueueFixture fix;
+    
+    MessageDataCollector collector;
+    for (uint i = 0; i < fix.queues.size(); i++) {
+        Receiver r = fix.session.createReceiver(fix.queues[i]);
+        r.setListener(&collector);
+        r.setCapacity(10u);
+        r.start();//TODO: add Session::start
+    }
+
+    for (uint i = 0; i < fix.queues.size(); i++) {
+        Sender s = fix.session.createSender(fix.queues[i]);
+        Message msg((boost::format("Message_%1%") % (i+1)).str());
+        s.send(msg);
+    }    
+
+    while (fix.session.dispatch(qpid::sys::TIME_SEC)) ;
+    
+    BOOST_CHECK_EQUAL(collector.messageData, boost::assign::list_of<std::string>("Message_1")("Message_2")("Message_3"));
+}
+
+
+QPID_AUTO_TEST_CASE(testMapMessage)
+{
+    QueueFixture fix;
+    Sender sender = fix.session.createSender(fix.queue);
+    Message out;
+    out.getContent().asMap()["abc"] = "def";
+    out.getContent().asMap()["pi"] = 3.14f;
+    sender.send(out);
+    Receiver receiver = fix.session.createReceiver(fix.queue);
+    Message in = receiver.fetch(5 * qpid::sys::TIME_SEC);    
+    BOOST_CHECK_EQUAL(in.getContent().asMap()["abc"].asString(), "def");
+    BOOST_CHECK_EQUAL(in.getContent().asMap()["pi"].asFloat(), 3.14f);
+    fix.session.acknowledge();
+}
+
+QPID_AUTO_TEST_CASE(testListMessage)
+{
+    QueueFixture fix;
+    Sender sender = fix.session.createSender(fix.queue);
+    Message out;
+    out.getContent() = Variant::List();
+    out.getContent() << "abc";
+    out.getContent() << 1234;
+    out.getContent() << "def";
+    out.getContent() << 56.789;    
+    sender.send(out);
+    Receiver receiver = fix.session.createReceiver(fix.queue);
+    Message in = receiver.fetch(5 * qpid::sys::TIME_SEC);    
+    Variant::List& list = in.getContent().asList();    
+    BOOST_CHECK_EQUAL(list.size(), out.getContent().asList().size());
+    BOOST_CHECK_EQUAL(list.front().asString(), "abc");
+    list.pop_front();
+    BOOST_CHECK_EQUAL(list.front().asInt64(), 1234);
+    list.pop_front();
+    BOOST_CHECK_EQUAL(list.front().asString(), "def");
+    list.pop_front();
+    BOOST_CHECK_EQUAL(list.front().asDouble(), 56.789);
+    fix.session.acknowledge();
+}
+
+QPID_AUTO_TEST_CASE(testReject)
+{
+    QueueFixture fix;
+    Sender sender = fix.session.createSender(fix.queue);
+    Message m1("reject-me");
+    sender.send(m1);
+    Message m2("accept-me");
+    sender.send(m2);
+    Receiver receiver = fix.session.createReceiver(fix.queue);
+    Message in = receiver.fetch(5 * qpid::sys::TIME_SEC);
+    BOOST_CHECK_EQUAL(in.getBytes(), m1.getBytes());
+    fix.session.reject(in);
+    in = receiver.fetch(5 * qpid::sys::TIME_SEC);
+    BOOST_CHECK_EQUAL(in.getBytes(), m2.getBytes());
+    fix.session.acknowledge();    
+}
+
+QPID_AUTO_TEST_SUITE_END()
diff --git a/qpid/cpp/src/tests/Variant.cpp b/qpid/cpp/src/tests/Variant.cpp
new file mode 100644
index 0000000000..b7ce776827
--- /dev/null
+++ b/qpid/cpp/src/tests/Variant.cpp
@@ -0,0 +1,157 @@
+/*
+ *
+ * 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 <iostream>
+#include "qpid/messaging/Variant.h"
+
+#include "unit_test.h"
+
+using namespace qpid::messaging;
+
+QPID_AUTO_TEST_SUITE(VariantSuite)
+
+QPID_AUTO_TEST_CASE(testConversions)
+{
+    Variant value;
+
+    //string to float/double
+    value = "1.5";
+    BOOST_CHECK_EQUAL((float) 1.5, value.asFloat());
+    BOOST_CHECK_EQUAL((double) 1.5, value.asDouble());
+
+    //float to string or double
+    value = 1.5f;
+    BOOST_CHECK_EQUAL((float) 1.5, value.asFloat());
+    BOOST_CHECK_EQUAL((double) 1.5, value.asDouble());
+    BOOST_CHECK_EQUAL(std::string("1.5"), value.asString());
+
+    //double to string (conversion to float not valid)
+    value = 1.5;
+    BOOST_CHECK_THROW(value.asFloat(), InvalidConversion);
+    BOOST_CHECK_EQUAL((double) 1.5, value.asDouble());
+    BOOST_CHECK_EQUAL(std::string("1.5"), value.asString());
+
+    //uint8 to larger unsigned ints and string
+    value = (uint8_t) 7;
+    BOOST_CHECK_EQUAL((uint8_t) 7, value.asUint8());
+    BOOST_CHECK_EQUAL((uint16_t) 7, value.asUint16());
+    BOOST_CHECK_EQUAL((uint32_t) 7, value.asUint32());
+    BOOST_CHECK_EQUAL((uint64_t) 7, value.asUint64());
+    BOOST_CHECK_EQUAL(std::string("7"), value.asString());
+    BOOST_CHECK_THROW(value.asInt8(), InvalidConversion);
+
+    value = (uint16_t) 8;
+    BOOST_CHECK_EQUAL(std::string("8"), value.asString());
+    value = (uint32_t) 9;
+    BOOST_CHECK_EQUAL(std::string("9"), value.asString());
+
+    //uint32 to larger unsigned ints and string
+    value = (uint32_t) 9999999;
+    BOOST_CHECK_EQUAL((uint32_t) 9999999, value.asUint32());
+    BOOST_CHECK_EQUAL((uint64_t) 9999999, value.asUint64());
+    BOOST_CHECK_EQUAL(std::string("9999999"), value.asString());
+    BOOST_CHECK_THROW(value.asUint8(), InvalidConversion);
+    BOOST_CHECK_THROW(value.asUint16(), InvalidConversion);
+    BOOST_CHECK_THROW(value.asInt32(), InvalidConversion);
+
+    value = "true";
+    BOOST_CHECK(value.asBool());
+    value = "false";
+    BOOST_CHECK(!value.asBool());
+    value = "1";
+    BOOST_CHECK(value.asBool());
+    value = "0";
+    BOOST_CHECK(!value.asBool());
+    value = "other";
+    BOOST_CHECK_THROW(value.asBool(), InvalidConversion);
+}
+
+QPID_AUTO_TEST_CASE(testAssignment)
+{
+    Variant value("abc");
+    Variant other = value;
+    BOOST_CHECK_EQUAL(VAR_STRING, value.getType());
+    BOOST_CHECK_EQUAL(other.getType(), value.getType());
+    BOOST_CHECK_EQUAL(other.asString(), value.asString());
+
+    const uint32_t i(1000);
+    value = i;
+    BOOST_CHECK_EQUAL(VAR_UINT32, value.getType());
+    BOOST_CHECK_EQUAL(VAR_STRING, other.getType());    
+}
+
+QPID_AUTO_TEST_CASE(testList)
+{    
+    const std::string s("abc");
+    const float f(9.876f);
+    const int16_t x(1000);
+
+    Variant value = Variant::List();
+    value.asList().push_back(Variant(s));
+    value.asList().push_back(Variant(f));
+    value.asList().push_back(Variant(x));
+    BOOST_CHECK_EQUAL(3u, value.asList().size());
+    Variant::List::const_iterator i = value.asList().begin(); 
+
+    BOOST_CHECK(i != value.asList().end());
+    BOOST_CHECK_EQUAL(VAR_STRING, i->getType());
+    BOOST_CHECK_EQUAL(s, i->asString());
+    i++;
+
+    BOOST_CHECK(i != value.asList().end());
+    BOOST_CHECK_EQUAL(VAR_FLOAT, i->getType());
+    BOOST_CHECK_EQUAL(f, i->asFloat());
+    i++;
+
+    BOOST_CHECK(i != value.asList().end());
+    BOOST_CHECK_EQUAL(VAR_INT16, i->getType());
+    BOOST_CHECK_EQUAL(x, i->asInt16());
+    i++;
+
+    BOOST_CHECK(i == value.asList().end());
+}
+
+QPID_AUTO_TEST_CASE(testMap)
+{    
+    const std::string red("red");
+    const float pi(3.14f);
+    const int16_t x(1000);
+
+    Variant value = Variant::Map();
+    value.asMap()["colour"] = red;
+    value.asMap()["pi"] = pi;
+    value.asMap()["my-key"] = x;
+    BOOST_CHECK_EQUAL(3u, value.asMap().size());
+
+    BOOST_CHECK_EQUAL(VAR_STRING, value.asMap()["colour"].getType());
+    BOOST_CHECK_EQUAL(red, value.asMap()["colour"].asString());
+
+    BOOST_CHECK_EQUAL(VAR_FLOAT, value.asMap()["pi"].getType());
+    BOOST_CHECK_EQUAL(pi, value.asMap()["pi"].asFloat());
+    
+    BOOST_CHECK_EQUAL(VAR_INT16, value.asMap()["my-key"].getType());
+    BOOST_CHECK_EQUAL(x, value.asMap()["my-key"].asInt16());
+
+    value.asMap()["my-key"] = "now it's a string";
+    BOOST_CHECK_EQUAL(VAR_STRING, value.asMap()["my-key"].getType());
+    BOOST_CHECK_EQUAL(std::string("now it's a string"), value.asMap()["my-key"].asString());
+}
+ 
+QPID_AUTO_TEST_SUITE_END()
diff --git a/qpid/cpp/src/tests/cli_tests.py b/qpid/cpp/src/tests/cli_tests.py
index 2af1a20a87..4309b66271 100755
--- a/qpid/cpp/src/tests/cli_tests.py
+++ b/qpid/cpp/src/tests/cli_tests.py
@@ -123,6 +123,28 @@ class CliTests(TestBase010):
                 found = True
         self.assertEqual(found, False)
 
+    def test_qpid_config_altex(self):
+        self.startQmf();
+        qmf = self.qmf
+        exName = "testalt"
+        altName = "amq.direct"
+
+        ret = os.system(self.command(" add exchange topic %s --alternate-exchange=%s" % (exName, altName)))
+        self.assertEqual(ret, 0)
+
+        exchanges = qmf.getObjects(_class="exchange")
+        found = False
+        for exchange in exchanges:
+            if exchange.name == altName:
+                self.assertEqual(exchange.altExchange, None)
+
+            if exchange.name == exName:
+                found = True
+                if not exchange.altExchange:
+                    self.fail("Alternate exchange not set")
+                self.assertEqual(exchange._altExchange_.name, altName)
+        self.assertEqual(found, True)
+
     def test_qpid_route(self):
         self.startQmf();
         qmf = self.qmf
diff --git a/qpid/cpp/src/tests/cluster_python_tests_failing.txt b/qpid/cpp/src/tests/cluster_python_tests_failing.txt
index 337fb4a0f2..53b609942d 100644
--- a/qpid/cpp/src/tests/cluster_python_tests_failing.txt
+++ b/qpid/cpp/src/tests/cluster_python_tests_failing.txt
@@ -1,4 +1,5 @@
 tests_0-10.management.ManagementTest.test_purge_queue
+tests_0-10.management.ManagementTest.test_connection_close
 tests_0-10.dtx.DtxTests.test_bad_resume
 tests_0-10.dtx.DtxTests.test_commit_unknown
 tests_0-10.dtx.DtxTests.test_end
diff --git a/qpid/cpp/src/tests/federation.py b/qpid/cpp/src/tests/federation.py
index daf831f3ed..aa68e8198b 100755
--- a/qpid/cpp/src/tests/federation.py
+++ b/qpid/cpp/src/tests/federation.py
@@ -32,6 +32,17 @@ class FederationTests(TestBase010):
     def remote_port(self):
         return int(self.defines["remote-port"])
 
+    def verify_cleanup(self):
+        attempts = 0
+        total = len(self.qmf.getObjects(_class="bridge")) + len(self.qmf.getObjects(_class="link"))
+        while total > 0:
+            attempts += 1
+            if attempts >= 10:
+                self.fail("Bridges and links didn't clean up")
+                return
+            sleep(1)
+            total = len(self.qmf.getObjects(_class="bridge")) + len(self.qmf.getObjects(_class="link"))
+
     def test_bridge_create_and_close(self):
         self.startQmf();
         qmf = self.qmf
@@ -51,9 +62,7 @@ class FederationTests(TestBase010):
         result = link.close()
         self.assertEqual(result.status, 0)
 
-        sleep(3)
-        self.assertEqual(len(qmf.getObjects(_class="bridge")), 0)
-        self.assertEqual(len(qmf.getObjects(_class="link")), 0)
+        self.verify_cleanup()
 
     def test_pull_from_exchange(self):
         session = self.session
@@ -98,9 +107,7 @@ class FederationTests(TestBase010):
         result = link.close()
         self.assertEqual(result.status, 0)
 
-        sleep(3)
-        self.assertEqual(len(qmf.getObjects(_class="bridge")), 0)
-        self.assertEqual(len(qmf.getObjects(_class="link")), 0)
+        self.verify_cleanup()
 
     def test_push_to_exchange(self):
         session = self.session
@@ -144,9 +151,7 @@ class FederationTests(TestBase010):
         result = link.close()
         self.assertEqual(result.status, 0)
 
-        sleep(3)
-        self.assertEqual(len(qmf.getObjects(_class="bridge")), 0)
-        self.assertEqual(len(qmf.getObjects(_class="link")), 0)
+        self.verify_cleanup()
 
     def test_pull_from_queue(self):
         session = self.session
@@ -199,9 +204,7 @@ class FederationTests(TestBase010):
         result = link.close()
         self.assertEqual(result.status, 0)
 
-        sleep(3)
-        self.assertEqual(len(qmf.getObjects(_class="bridge")), 0)
-        self.assertEqual(len(qmf.getObjects(_class="link")), 0)
+        self.verify_cleanup()
 
     def test_tracing_automatic(self):
         remoteUrl = "%s:%d" % (self.remote_host(), self.remote_port())
@@ -312,9 +315,7 @@ class FederationTests(TestBase010):
         result = link.close()
         self.assertEqual(result.status, 0)
 
-        sleep(3)
-        self.assertEqual(len(qmf.getObjects(_class="bridge")), 0)
-        self.assertEqual(len(qmf.getObjects(_class="link")), 0)
+        self.verify_cleanup()
 
     def test_dynamic_fanout(self):
         session = self.session
@@ -358,9 +359,7 @@ class FederationTests(TestBase010):
         result = link.close()
         self.assertEqual(result.status, 0)
 
-        sleep(3)
-        self.assertEqual(len(qmf.getObjects(_class="bridge")), 0)
-        self.assertEqual(len(qmf.getObjects(_class="link")), 0)
+        self.verify_cleanup()
 
 
     def test_dynamic_direct(self):
@@ -405,10 +404,7 @@ class FederationTests(TestBase010):
         result = link.close()
         self.assertEqual(result.status, 0)
 
-        sleep(3)
-        self.assertEqual(len(qmf.getObjects(_class="bridge")), 0)
-        self.assertEqual(len(qmf.getObjects(_class="link")), 0)
-
+        self.verify_cleanup()
 
     def test_dynamic_topic(self):
         session = self.session
@@ -452,9 +448,7 @@ class FederationTests(TestBase010):
         result = link.close()
         self.assertEqual(result.status, 0)
 
-        sleep(3)
-        self.assertEqual(len(qmf.getObjects(_class="bridge")), 0)
-        self.assertEqual(len(qmf.getObjects(_class="link")), 0)
+        self.verify_cleanup()
 
     def test_dynamic_topic_reorigin(self):
         session = self.session
@@ -509,14 +503,24 @@ class FederationTests(TestBase010):
         self.assertEqual(result.status, 0)
         result = bridge2.close()
         self.assertEqual(result.status, 0)
-        result = link.close()
-        self.assertEqual(result.status, 0)
 
-        sleep(3)
-        self.assertEqual(len(qmf.getObjects(_class="bridge")), 0)
-        self.assertEqual(len(qmf.getObjects(_class="link")), 0)
+        # extra check: verify we don't leak bridge objects - keep the link
+        # around and verify the bridge count has gone to zero
+
+        attempts = 0
+        bridgeCount = len(qmf.getObjects(_class="bridge"))
+        while bridgeCount > 0:
+            attempts += 1
+            if attempts >= 5:
+                self.fail("Bridges didn't clean up")
+                return
+            sleep(1)
+            bridgeCount = len(qmf.getObjects(_class="bridge"))
 
+        result = link.close()
+        self.assertEqual(result.status, 0)
 
+        self.verify_cleanup()
         
     def test_dynamic_direct_reorigin(self):
         session = self.session
@@ -569,16 +573,17 @@ class FederationTests(TestBase010):
 
         result = bridge.close()
         self.assertEqual(result.status, 0)
-        result = bridge2.close()
-        self.assertEqual(result.status, 0)
+        
+        # Extra test: don't explicitly close() bridge2.  When the link is closed,
+        # it should clean up bridge2 automagically.  verify_cleanup() will detect
+        # if bridge2 isn't cleaned up and will fail the test.
+        #
+        #result = bridge2.close()
+        #self.assertEqual(result.status, 0)
         result = link.close()
         self.assertEqual(result.status, 0)
 
-        sleep(3)
-        self.assertEqual(len(qmf.getObjects(_class="bridge")), 0)
-        self.assertEqual(len(qmf.getObjects(_class="link")), 0)
-
-
+        self.verify_cleanup()
 
     def getProperty(self, msg, name):
         for h in msg.headers:
diff --git a/qpid/cpp/src/tests/qpid_stream.cpp b/qpid/cpp/src/tests/qpid_stream.cpp
new file mode 100644
index 0000000000..8e02baa8a0
--- /dev/null
+++ b/qpid/cpp/src/tests/qpid_stream.cpp
@@ -0,0 +1,163 @@
+/*
+ *
+ * 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/messaging/Connection.h>
+#include <qpid/messaging/Message.h>
+#include <qpid/messaging/Receiver.h>
+#include <qpid/messaging/Sender.h>
+#include <qpid/messaging/Session.h>
+#include <qpid/sys/Runnable.h>
+#include <qpid/sys/Thread.h>
+#include <qpid/sys/Time.h>
+#include <qpid/Options.h>
+#include <iostream>
+#include <string>
+
+using namespace qpid::messaging;
+using namespace qpid::sys;
+
+struct Args : public qpid::Options 
+{
+    std::string url;
+    std::string address;
+    uint rate;
+    bool durable;
+
+    Args() : url("amqp:tcp:127.0.0.1:5672"), address("test-queue"), rate(1000), durable(false)
+    {
+        addOptions()
+            ("url", qpid::optValue(url, "URL"), "Url to connect to.")
+            ("address", qpid::optValue(address, "ADDRESS"), "Address to stream messages through.")
+            ("rate", qpid::optValue(rate, "msgs/sec"), "Rate at which to stream messages.")
+            ("durable", qpid::optValue(durable, "true|false"), "Mark messages as durable.");
+    }
+};
+
+Args opts;
+
+const std::string TIMESTAMP = "ts";
+
+uint64_t timestamp(const AbsTime& time)
+{
+    Duration t(time);
+    return t;
+}
+
+struct Client : Runnable
+{
+    virtual ~Client() {}
+    virtual void doWork(Session&) = 0;
+
+    void run()
+    {
+        try {
+            Connection connection = Connection::open(opts.url);
+            Session session = connection.newSession();
+            doWork(session);
+            session.close();
+            connection.close();
+        } catch(const std::exception& error) {
+            std::cout << error.what() << std::endl;
+        }
+    }
+
+    Thread thread;
+
+    void start() { thread = Thread(this); }
+    void join() { thread.join(); }
+};
+
+struct Publish : Client
+{
+    void doWork(Session& session)
+    {
+        Sender sender = session.createSender(opts.address);
+        Message msg;
+        uint64_t interval = TIME_SEC / opts.rate;
+        uint64_t sent = 0, missedRate = 0;
+        AbsTime start = now();
+        while (true) {
+            AbsTime sentAt = now();
+            msg.getHeaders()[TIMESTAMP] = timestamp(sentAt);
+            sender.send(msg);
+            ++sent;
+            AbsTime waitTill(start, sent*interval);
+            Duration delay(sentAt, waitTill);
+            if (delay < 0) {
+                ++missedRate;
+            } else {
+                qpid::sys::usleep(delay / TIME_USEC);
+            }
+        }
+    }
+};
+
+struct Consume : Client
+{
+    void doWork(Session& session)
+    {
+        Message msg;
+        uint64_t received = 0;
+        double minLatency = std::numeric_limits<double>::max();
+        double maxLatency = 0;
+        double totalLatency = 0;
+        Receiver receiver = session.createReceiver(opts.address);
+        while (receiver.fetch(msg)) {
+            session.acknowledge();//TODO: add batching option
+            ++received;
+            //calculate latency
+            uint64_t receivedAt = timestamp(now());
+            uint64_t sentAt = msg.getHeaders()[TIMESTAMP].asUint64();
+            double latency = ((double) (receivedAt - sentAt)) / TIME_MSEC;
+
+            //update avg, min & max
+            minLatency = std::min(minLatency, latency);
+            maxLatency = std::max(maxLatency, latency);
+            totalLatency += latency;
+
+            if (received % opts.rate == 0) {
+                std::cout << "count=" << received 
+                          << ", avg=" << (totalLatency/received) 
+                          << ", min=" << minLatency 
+                          << ", max=" << maxLatency << std::endl;
+            }
+        }        
+    }
+};
+
+int main(int argc, char** argv)
+{
+    try {
+        opts.parse(argc, argv);
+        Publish publish;
+        Consume consume;
+        publish.start();
+        consume.start();
+        consume.join();
+        publish.join();
+        return 0;
+    } catch(const std::exception& error) {
+        std::cout << error.what() << std::endl;
+    }
+    return 1;
+}
+
+
diff --git a/qpid/cpp/src/tests/txtest.cpp b/qpid/cpp/src/tests/txtest.cpp
index c1ee246e2c..f604df7e21 100644
--- a/qpid/cpp/src/tests/txtest.cpp
+++ b/qpid/cpp/src/tests/txtest.cpp
@@ -33,7 +33,7 @@
 #include "qpid/client/SubscriptionManager.h"
 #include "qpid/framing/Array.h"
 #include "qpid/framing/Buffer.h"
-#include "qpid/sys/uuid.h"
+#include "qpid/framing/Uuid.h"
 #include "qpid/sys/Thread.h"
 
 using namespace qpid;
@@ -130,8 +130,6 @@ struct Transfer : public Client, public Runnable
     std::string src;
     std::string dest;
     Thread thread;
-    uuid_t uuid;
-    char uuidStr[37]; // Format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx + trailing \0
     framing::Xid xid;
 
     Transfer(const std::string& to, const std::string& from) : src(to), dest(from), xid(0x4c414e47, "", from) {}
@@ -184,9 +182,8 @@ struct Transfer : public Client, public Runnable
     }
 
     void setNewXid(framing::Xid& xid) {
-        ::uuid_generate(uuid);
-        ::uuid_unparse(uuid, uuidStr);
-        xid.setGlobalId(uuidStr);
+        framing::Uuid uuid(true);
+        xid.setGlobalId(uuid.str());
     }
 };