QPID-664: Initial checkin of high level messaging api for c++

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

4 files changed, 487 insertions, 1 deletions

diff --git a/qpid/cpp/src/tests/CMakeLists.txt b/qpid/cpp/src/tests/CMakeLists.txt
index 34f5d35a9a..56a4aaf2b0 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
+    MessagingSessionTest
     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/Makefile.am b/qpid/cpp/src/tests/Makefile.am
index db4c8ba914..2e04c85b93 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
diff --git a/qpid/cpp/src/tests/MessagingSessionTests.cpp b/qpid/cpp/src/tests/MessagingSessionTests.cpp
new file mode 100644
index 0000000000..ef320c3ae0
--- /dev/null
+++ b/qpid/cpp/src/tests/MessagingSessionTests.cpp
@@ -0,0 +1,325 @@
+/*
+ *
+ * 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(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.getBytes(), out.getBytes());
+    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);    
+    BOOST_CHECK_EQUAL(in.getBytes(), out.getBytes());
+    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_SUITE_END()
diff --git a/qpid/cpp/src/tests/Variant.cpp b/qpid/cpp/src/tests/Variant.cpp
new file mode 100644
index 0000000000..1bf2ed98ce
--- /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(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(UINT32, value.getType());
+    BOOST_CHECK_EQUAL(STRING, other.getType());    
+}
+
+QPID_AUTO_TEST_CASE(testList)
+{    
+    const std::string s("abc");
+    const float f(9.876);
+    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(STRING, i->getType());
+    BOOST_CHECK_EQUAL(s, i->asString());
+    i++;
+
+    BOOST_CHECK(i != value.asList().end());
+    BOOST_CHECK_EQUAL(FLOAT, i->getType());
+    BOOST_CHECK_EQUAL(f, i->asFloat());
+    i++;
+
+    BOOST_CHECK(i != value.asList().end());
+    BOOST_CHECK_EQUAL(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.14);
+    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(STRING, value.asMap()["colour"].getType());
+    BOOST_CHECK_EQUAL(red, value.asMap()["colour"].asString());
+
+    BOOST_CHECK_EQUAL(FLOAT, value.asMap()["pi"].getType());
+    BOOST_CHECK_EQUAL(pi, value.asMap()["pi"].asFloat());
+    
+    BOOST_CHECK_EQUAL(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(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()