Added support for address parsing, create/assert/delete policies

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

3 files changed, 386 insertions, 11 deletions

diff --git a/cpp/src/tests/Address.cpp b/cpp/src/tests/Address.cpp
new file mode 100644
index 0000000000..ab4017a788
--- /dev/null
+++ b/cpp/src/tests/Address.cpp
@@ -0,0 +1,91 @@
+/*
+ *
+ * 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/Address.h"
+#include "qpid/messaging/Variant.h"
+
+#include "unit_test.h"
+
+using namespace qpid::messaging;
+
+namespace qpid {
+namespace tests {
+
+QPID_AUTO_TEST_SUITE(AddressSuite)
+
+QPID_AUTO_TEST_CASE(testParseNameOnly)
+{
+    Address address("my-topic");
+    BOOST_CHECK_EQUAL(std::string("my-topic"), address.getName());
+}
+
+QPID_AUTO_TEST_CASE(testParseSubject)
+{
+    Address address("my-topic/my-subject");
+    BOOST_CHECK_EQUAL(std::string("my-topic"), address.getName());
+    BOOST_CHECK_EQUAL(std::string("my-subject"), address.getSubject());
+}
+
+QPID_AUTO_TEST_CASE(testParseOptions)
+{
+    Address address("my-topic {a:bc, x:101, y:'a string'}");
+    BOOST_CHECK_EQUAL(std::string("my-topic"), address.getName());
+    BOOST_CHECK_EQUAL(std::string("bc"), address.getOption("a").asString());
+    BOOST_CHECK_EQUAL((uint16_t) 101, address.getOption("x").asInt64());
+    BOOST_CHECK_EQUAL(std::string("a string"), address.getOption("y").asString());
+}
+
+QPID_AUTO_TEST_CASE(testParseSubjectAndOptions)
+{
+    Address address("my-topic/my-subject {a:bc, x:101, y:'a string'}");
+    BOOST_CHECK_EQUAL(std::string("my-topic"), address.getName());
+    BOOST_CHECK_EQUAL(std::string("my-subject"), address.getSubject());
+    BOOST_CHECK_EQUAL(std::string("bc"), address.getOption("a").asString());
+    BOOST_CHECK_EQUAL((uint16_t) 101, address.getOption("x").asInt64());
+    BOOST_CHECK_EQUAL(std::string("a string"), address.getOption("y").asString());
+}
+
+QPID_AUTO_TEST_CASE(testParseNestedOptions)
+{
+    Address address("my-topic {a:{p:202, q:'another string'}, x:101, y:'a string'}");
+    BOOST_CHECK_EQUAL(std::string("my-topic"), address.getName());
+    BOOST_CHECK_EQUAL((uint16_t) 202, address.getOptions()["a"].asMap()["p"].asInt64());
+    BOOST_CHECK_EQUAL(std::string("another string"), address.getOptions()["a"].asMap()["q"].asString());
+    BOOST_CHECK_EQUAL((uint16_t) 101, address.getOption("x").asInt64());
+    BOOST_CHECK_EQUAL(std::string("a string"), address.getOption("y").asString());
+}
+
+QPID_AUTO_TEST_CASE(testParseOptionsWithList)
+{
+    Address address("my-topic {a:[202, 'another string'], x:101}");
+    BOOST_CHECK_EQUAL(std::string("my-topic"), address.getName());
+    Variant::List& list = address.getOptions()["a"].asList();
+    Variant::List::const_iterator i = list.begin();
+    BOOST_CHECK(i != list.end());
+    BOOST_CHECK_EQUAL((uint16_t) 202, i->asInt64());
+    BOOST_CHECK(++i != list.end());
+    BOOST_CHECK_EQUAL(std::string("another string"), i->asString());
+    BOOST_CHECK_EQUAL((uint16_t) 101, address.getOption("x").asInt64());
+}
+
+QPID_AUTO_TEST_SUITE_END()
+
+}}
diff --git a/cpp/src/tests/Makefile.am b/cpp/src/tests/Makefile.am
index e1935dfcc6..b84251831d 100644
--- a/cpp/src/tests/Makefile.am
+++ b/cpp/src/tests/Makefile.am
@@ -114,7 +114,8 @@ unit_test_SOURCES= unit_test.cpp unit_test.h \
 	ReplicationTest.cpp \
 	ClientMessageTest.cpp \
 	PollableCondition.cpp \
-	Variant.cpp
+	Variant.cpp \
+	Address.cpp
 
 if HAVE_XML
 unit_test_SOURCES+= XmlClientSessionTest.cpp
diff --git a/cpp/src/tests/MessagingSessionTests.cpp b/cpp/src/tests/MessagingSessionTests.cpp
index 206f5ba691..fc39557a0e 100644
--- a/cpp/src/tests/MessagingSessionTests.cpp
+++ b/cpp/src/tests/MessagingSessionTests.cpp
@@ -21,6 +21,7 @@
 #include "unit_test.h"
 #include "test_tools.h"
 #include "BrokerFixture.h"
+#include "qpid/messaging/Address.h"
 #include "qpid/messaging/Connection.h"
 #include "qpid/messaging/ListContent.h"
 #include "qpid/messaging/ListView.h"
@@ -33,7 +34,9 @@
 #include "qpid/messaging/Session.h"
 #include "qpid/client/Connection.h"
 #include "qpid/client/Session.h"
+#include "qpid/framing/ExchangeQueryResult.h"
 #include "qpid/framing/reply_exceptions.h"
+#include "qpid/framing/Uuid.h"
 #include "qpid/sys/Time.h"
 #include <boost/assign.hpp>
 #include <boost/format.hpp>
@@ -48,6 +51,7 @@ QPID_AUTO_TEST_SUITE(MessagingSessionTests)
 using namespace qpid::messaging;
 using namespace qpid;
 using qpid::broker::Broker;
+using qpid::framing::Uuid;
 
 struct BrokerAdmin
 {
@@ -80,6 +84,18 @@ struct BrokerAdmin
         session.exchangeDelete(qpid::client::arg::exchange=name);
     }
 
+    bool checkQueueExists(const std::string& name)
+    {
+        return session.queueQuery(name).getQueue() == name;
+    }
+
+    bool checkExchangeExists(const std::string& name, std::string& type)
+    {
+        qpid::framing::ExchangeQueryResult result = session.exchangeQuery(name);
+        type = result.getType();
+        return !result.getNotFound();
+    }
+
     ~BrokerAdmin()
     {
         session.close();
@@ -99,6 +115,19 @@ struct MessagingFixture : public BrokerFixture
         session(connection.newSession()),
         admin(broker->getPort(Broker::TCP_TRANSPORT)) {}
 
+    void ping(const qpid::messaging::Address& address)
+    {
+        Receiver r = session.createReceiver(address);
+        Sender s = session.createSender(address);
+        Message out(Uuid(true).str());
+        s.send(out);
+        Message in;
+        BOOST_CHECK(r.fetch(in, 5*qpid::sys::TIME_SEC));
+        BOOST_CHECK_EQUAL(out.getContent(), in.getContent());
+        r.cancel();
+        s.cancel();
+    }
+
     ~MessagingFixture()
     {
         session.close();
@@ -178,6 +207,22 @@ std::vector<std::string> fetch(Receiver& receiver, int count, qpid::sys::Duratio
     return data;
 }
 
+
+void send(Sender& sender, uint count = 1, uint start = 1, const std::string& base = "Message")
+{
+    for (uint i = start; i < start + count; ++i) {
+        sender.send(Message((boost::format("%1%_%2%") % base % i).str()));
+    }
+}
+
+void receive(Receiver& receiver, uint count = 1, uint start = 1,
+             const std::string& base = "Message", qpid::sys::Duration timeout=qpid::sys::TIME_SEC*5)
+{
+    for (uint i = start; i < start + count; ++i) {
+        BOOST_CHECK_EQUAL(receiver.fetch(timeout).getContent(), (boost::format("%1%_%2%") % base % i).str());
+    }
+}
+
 QPID_AUTO_TEST_CASE(testSimpleSendReceive)
 {
     QueueFixture fix;
@@ -212,15 +257,19 @@ QPID_AUTO_TEST_CASE(testSendReceiveHeaders)
 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);
+    ScopedSuppressLogging sl;
+    BOOST_CHECK_THROW(fix.session.createSender("NonExistentAddress"), qpid::messaging::InvalidAddress);
+    BOOST_CHECK_THROW(fix.session.createSender("NonExistentAddress {create:receiver, type:queue}"),
+                      qpid::messaging::InvalidAddress);
 }
 
 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);
+    ScopedSuppressLogging sl;
+    BOOST_CHECK_THROW(fix.session.createReceiver("NonExistentAddress"), qpid::messaging::InvalidAddress);
+    BOOST_CHECK_THROW(fix.session.createReceiver("NonExistentAddress {create:sender, type:queue}"),
+                      qpid::messaging::InvalidAddress);
 }
 
 QPID_AUTO_TEST_CASE(testSimpleTopic)
@@ -433,9 +482,7 @@ QPID_AUTO_TEST_CASE(testPendingSend)
 {
     QueueFixture fix;
     Sender sender = fix.session.createSender(fix.queue);
-    for (uint i = 0; i < 10; ++i) {
-        sender.send(Message((boost::format("Message_%1%") % (i+1)).str()));
-    }
+    send(sender, 10);
     //Note: this test relies on 'inside knowledge' of the sender
     //implementation and the fact that the simple test case makes it
     //possible to predict when completion information will be sent to
@@ -445,12 +492,248 @@ QPID_AUTO_TEST_CASE(testPendingSend)
     BOOST_CHECK_EQUAL(sender.pending(), 0u);
 
     Receiver receiver = fix.session.createReceiver(fix.queue);
-    for (uint i = 0; i < 10; ++i) {
-        BOOST_CHECK_EQUAL(receiver.fetch().getContent(), (boost::format("Message_%1%") % (i+1)).str());
-    }
+    receive(receiver, 10);
+    fix.session.acknowledge();
+}
+
+QPID_AUTO_TEST_CASE(testBrowse)
+{
+    QueueFixture fix;
+    Sender sender = fix.session.createSender(fix.queue);
+    send(sender, 10);
+    Receiver browser1 = fix.session.createReceiver(fix.queue + " {browse:true}");
+    receive(browser1, 10);
+    Receiver browser2 = fix.session.createReceiver(fix.queue + " {browse:true}");
+    receive(browser2, 10);
+    Receiver consumer = fix.session.createReceiver(fix.queue);
+    receive(consumer, 10);
     fix.session.acknowledge();
 }
 
+struct QueueCreatePolicyFixture : public MessagingFixture
+{
+    qpid::messaging::Address address;
+
+    QueueCreatePolicyFixture(const std::string& a) : address(a) {}
+
+    void test()
+    {
+        ping(address);
+        BOOST_CHECK(admin.checkQueueExists(address.getName()));
+    }
+
+    ~QueueCreatePolicyFixture()
+    {
+        admin.deleteQueue(address.getName());    
+    }
+};
+
+QPID_AUTO_TEST_CASE(testCreatePolicyQueueAlways)
+{
+    QueueCreatePolicyFixture fix("# {create:always, type:queue}");
+    fix.test();
+}
+
+QPID_AUTO_TEST_CASE(testCreatePolicyQueueReceiver)
+{
+    QueueCreatePolicyFixture fix("# {create:receiver, type:queue}");
+    Receiver r = fix.session.createReceiver(fix.address);
+    fix.test();
+    r.cancel();
+}
+
+QPID_AUTO_TEST_CASE(testCreatePolicyQueueSender)
+{
+    QueueCreatePolicyFixture fix("# {create:sender, type:queue}");
+    Sender s = fix.session.createSender(fix.address);
+    fix.test();
+    s.cancel();
+}
+
+struct ExchangeCreatePolicyFixture : public MessagingFixture
+{
+    qpid::messaging::Address address;
+    const std::string exchangeType;
+
+    ExchangeCreatePolicyFixture(const std::string& a, const std::string& t) :
+        address(a), exchangeType(t) {}
+
+    void test()
+    {
+        ping(address);
+        std::string actualType;
+        BOOST_CHECK(admin.checkExchangeExists(address.getName(), actualType));
+        BOOST_CHECK_EQUAL(exchangeType, actualType);
+    }
+
+    ~ExchangeCreatePolicyFixture()
+    {
+        admin.deleteExchange(address.getName());    
+    }
+};
+
+QPID_AUTO_TEST_CASE(testCreatePolicyTopic)
+{
+    ExchangeCreatePolicyFixture fix("# {create:always, type:topic, node-properties:{x-amqp0-10-exchange-type:topic}}",
+                                  "topic");
+    fix.test();
+}
+
+QPID_AUTO_TEST_CASE(testCreatePolicyTopicReceiverFanout)
+{
+    ExchangeCreatePolicyFixture fix("#/my-subject {create:receiver, type:topic, node-properties:{x-amqp0-10-exchange-type:fanout}}", "fanout");
+    Receiver r = fix.session.createReceiver(fix.address);
+    fix.test();
+    r.cancel();
+}
+
+QPID_AUTO_TEST_CASE(testCreatePolicyTopicSenderDirect)
+{
+    ExchangeCreatePolicyFixture fix("#/my-subject {create:sender, type:topic, node-properties:{x-amqp0-10-exchange-type:direct}}", "direct");
+    Sender s = fix.session.createSender(fix.address);
+    fix.test();
+    s.cancel();
+}
+
+struct DeletePolicyFixture : public MessagingFixture
+{
+    enum Mode {RECEIVER, SENDER, ALWAYS, NEVER};
+
+    std::string getPolicy(Mode mode)
+    {
+        switch (mode) {
+          case SENDER:
+            return "{delete:sender}";
+          case RECEIVER:
+            return "{delete:receiver}";
+          case ALWAYS:
+            return "{delete:always}";
+          case NEVER:
+            return "{delete:never}";
+        }
+    }
+
+    void testAll()
+    {
+        test(RECEIVER);
+        test(SENDER);
+        test(ALWAYS);
+        test(NEVER);
+    }
+
+    virtual ~DeletePolicyFixture() {}
+    virtual void create(const qpid::messaging::Address&) = 0;
+    virtual void destroy(const qpid::messaging::Address&) = 0;
+    virtual bool exists(const qpid::messaging::Address&) = 0;
+
+    void test(Mode mode)
+    {
+        qpid::messaging::Address address("# " + getPolicy(mode));
+        create(address);
+
+        Sender s = session.createSender(address);
+        Receiver r = session.createReceiver(address);
+        switch (mode) {
+          case RECEIVER:
+            s.cancel();
+            BOOST_CHECK(exists(address));
+            r.cancel();
+            BOOST_CHECK(!exists(address));
+            break;
+          case SENDER:
+            r.cancel();
+            BOOST_CHECK(exists(address));
+            s.cancel();
+            BOOST_CHECK(!exists(address));
+            break;
+          case ALWAYS:
+            //Problematic case at present; multiple attempts to delete
+            //will result in all but one attempt failing and killing
+            //the session which is not desirable. TODO: better
+            //implementation of delete policy.
+            s.cancel();
+            BOOST_CHECK(!exists(address));
+            break;
+          case NEVER:
+            r.cancel();
+            BOOST_CHECK(exists(address));
+            s.cancel();
+            BOOST_CHECK(exists(address));
+            destroy(address);
+        }
+    }
+};
+
+struct QueueDeletePolicyFixture : DeletePolicyFixture
+{
+    void create(const qpid::messaging::Address& address)
+    {
+        admin.createQueue(address.getName());
+    }
+    void destroy(const qpid::messaging::Address& address)
+    {
+        admin.deleteQueue(address.getName());
+    }
+    bool exists(const qpid::messaging::Address& address)
+    {
+        return admin.checkQueueExists(address.getName());
+    }
+};
+
+struct ExchangeDeletePolicyFixture : DeletePolicyFixture
+{
+    const std::string exchangeType;
+    ExchangeDeletePolicyFixture(const std::string type = "topic") : exchangeType(type) {}
+
+    void create(const qpid::messaging::Address& address)
+    {
+        admin.createExchange(address.getName(), exchangeType);
+    }
+    void destroy(const qpid::messaging::Address& address)
+    {
+        admin.deleteExchange(address.getName());
+    }
+    bool exists(const qpid::messaging::Address& address)
+    {
+        std::string actualType;
+        return admin.checkExchangeExists(address.getName(), actualType) && actualType == exchangeType;
+    }
+};
+
+QPID_AUTO_TEST_CASE(testDeletePolicyQueue)
+{
+    QueueDeletePolicyFixture fix;
+    fix.testAll();
+}
+
+QPID_AUTO_TEST_CASE(testDeletePolicyExchange)
+{
+    ExchangeDeletePolicyFixture fix;
+    fix.testAll();
+}
+
+QPID_AUTO_TEST_CASE(testAssertPolicyQueue)
+{
+    MessagingFixture fix;
+    std::string a1 = "q {create:always, assert:always, type:queue, node-properties:{durable:false, x-amqp0-10-arguments:{qpid.max-count:100}}}";
+    Sender s1 = fix.session.createSender(a1);
+    s1.cancel();
+    Receiver r1 = fix.session.createReceiver(a1);
+    r1.cancel();
+    
+    std::string a2 = "q {assert:receiver, node-properties:{durable:true, x-amqp0-10-arguments:{qpid.max-count:100}}}";
+    Sender s2 = fix.session.createSender(a2);
+    s2.cancel();
+    BOOST_CHECK_THROW(fix.session.createReceiver(a2), qpid::messaging::InvalidAddress);
+
+    std::string a3 = "q {assert:sender, node-properties:{x-amqp0-10-arguments:{qpid.max-count:99}}}";
+    BOOST_CHECK_THROW(fix.session.createSender(a3), qpid::messaging::InvalidAddress);
+    Receiver r3 = fix.session.createReceiver(a3);
+    r3.cancel();
+
+    fix.admin.deleteQueue("q");
+}
+
 QPID_AUTO_TEST_SUITE_END()
 
 }} // namespace qpid::tests