summaryrefslogtreecommitdiff
path: root/cpp/src/tests
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/tests')
-rw-r--r--cpp/src/tests/MessagingFixture.h117
-rw-r--r--cpp/src/tests/MessagingSessionTests.cpp47
-rw-r--r--cpp/src/tests/QueueTest.cpp2
-rwxr-xr-xcpp/src/tests/qpid-ctrl5
-rw-r--r--cpp/src/tests/sender.cpp2
5 files changed, 170 insertions, 3 deletions
diff --git a/cpp/src/tests/MessagingFixture.h b/cpp/src/tests/MessagingFixture.h
index 715de09bad..2312a87e9d 100644
--- a/cpp/src/tests/MessagingFixture.h
+++ b/cpp/src/tests/MessagingFixture.h
@@ -27,15 +27,19 @@
#include "qpid/client/Connection.h"
#include "qpid/client/Session.h"
#include "qpid/framing/Uuid.h"
+#include "qpid/messaging/Address.h"
#include "qpid/messaging/Connection.h"
#include "qpid/messaging/Session.h"
#include "qpid/messaging/Sender.h"
#include "qpid/messaging/Receiver.h"
#include "qpid/messaging/Message.h"
+#include "qpid/types/Variant.h"
namespace qpid {
namespace tests {
+using qpid::types::Variant;
+
struct BrokerAdmin
{
qpid::client::Connection connection;
@@ -223,6 +227,119 @@ inline void receive(messaging::Receiver& receiver, uint count = 1, uint start =
}
}
+
+class MethodInvoker
+{
+ public:
+ MethodInvoker(messaging::Session& session) : replyTo("#; {create:always, node:{x-declare:{auto-delete:true}}}"),
+ sender(session.createSender("qmf.default.direct/broker")),
+ receiver(session.createReceiver(replyTo)) {}
+
+ void createExchange(const std::string& name, const std::string& type, bool durable=false)
+ {
+ Variant::Map params;
+ params["name"]=name;
+ params["type"]="exchange";
+ params["properties"] = Variant::Map();
+ params["properties"].asMap()["exchange-type"] = type;
+ params["properties"].asMap()["durable"] = durable;
+ methodRequest("create", params);
+ }
+
+ void deleteExchange(const std::string& name)
+ {
+ Variant::Map params;
+ params["name"]=name;
+ params["type"]="exchange";
+ methodRequest("delete", params);
+ }
+
+ void createQueue(const std::string& name, bool durable=false, bool autodelete=false,
+ const Variant::Map& options=Variant::Map())
+ {
+ Variant::Map params;
+ params["name"]=name;
+ params["type"]="queue";
+ params["properties"] = options;
+ params["properties"].asMap()["durable"] = durable;
+ params["properties"].asMap()["auto-delete"] = autodelete;
+ methodRequest("create", params);
+ }
+
+ void deleteQueue(const std::string& name)
+ {
+ Variant::Map params;
+ params["name"]=name;
+ params["type"]="queue";
+ methodRequest("delete", params);
+ }
+
+ void bind(const std::string& exchange, const std::string& queue, const std::string& key,
+ const Variant::Map& options=Variant::Map())
+ {
+ Variant::Map params;
+ params["name"]=(boost::format("%1%/%2%/%3%") % (exchange) % (queue) % (key)).str();
+ params["type"]="binding";
+ params["properties"] = options;
+ methodRequest("create", params);
+ }
+
+ void unbind(const std::string& exchange, const std::string& queue, const std::string& key)
+ {
+ Variant::Map params;
+ params["name"]=(boost::format("%1%/%2%/%3%") % (exchange) % (queue) % (key)).str();
+ params["type"]="binding";
+ methodRequest("delete", params);
+ }
+
+ void methodRequest(const std::string& method, const Variant::Map& inParams, Variant::Map* outParams = 0)
+ {
+ Variant::Map content;
+ Variant::Map objectId;
+ objectId["_object_name"] = "org.apache.qpid.broker:broker:amqp-broker";
+ content["_object_id"] = objectId;
+ content["_method_name"] = method;
+ content["_arguments"] = inParams;
+
+ messaging::Message request;
+ request.setReplyTo(replyTo);
+ request.getProperties()["x-amqp-0-10.app-id"] = "qmf2";
+ request.getProperties()["qmf.opcode"] = "_method_request";
+ encode(content, request);
+
+ sender.send(request);
+
+ messaging::Message response;
+ if (receiver.fetch(response, messaging::Duration::SECOND*5)) {
+ if (response.getProperties()["x-amqp-0-10.app-id"] == "qmf2") {
+ std::string opcode = response.getProperties()["qmf.opcode"];
+ if (opcode == "_method_response") {
+ if (outParams) {
+ Variant::Map m;
+ decode(response, m);
+ *outParams = m["_arguments"].asMap();
+ }
+ } else if (opcode == "_exception") {
+ Variant::Map m;
+ decode(response, m);
+ throw Exception(QPID_MSG("Error: " << m["_values"]));
+ } else {
+ throw Exception(QPID_MSG("Invalid response received, unexpected opcode: " << opcode));
+ }
+ } else {
+ throw Exception(QPID_MSG("Invalid response received, not a qmfv2 message: app-id="
+ << response.getProperties()["x-amqp-0-10.app-id"]));
+ }
+ } else {
+ throw Exception(QPID_MSG("No response received"));
+ }
+ }
+ private:
+ messaging::Address replyTo;
+ messaging::Sender sender;
+ messaging::Receiver receiver;
+};
+
}} // namespace qpid::tests
#endif /*!TESTS_MESSAGINGFIXTURE_H*/
diff --git a/cpp/src/tests/MessagingSessionTests.cpp b/cpp/src/tests/MessagingSessionTests.cpp
index 991ec847bf..f9a8b0e4c1 100644
--- a/cpp/src/tests/MessagingSessionTests.cpp
+++ b/cpp/src/tests/MessagingSessionTests.cpp
@@ -890,6 +890,53 @@ QPID_AUTO_TEST_CASE(testAcknowledge)
BOOST_CHECK(!fix.session.createReceiver(fix.queue).fetch(m, Duration::IMMEDIATE));
}
+QPID_AUTO_TEST_CASE(testQmfCreateAndDelete)
+{
+ MessagingFixture fix(Broker::Options(), true/*enable management*/);
+ MethodInvoker control(fix.session);
+ control.createQueue("my-queue");
+ control.createExchange("my-exchange", "topic");
+ control.bind("my-exchange", "my-queue", "subject1");
+
+ Sender sender = fix.session.createSender("my-exchange");
+ Receiver receiver = fix.session.createReceiver("my-queue");
+ Message out;
+ out.setSubject("subject1");
+ out.setContent("one");
+ sender.send(out);
+ Message in;
+ BOOST_CHECK(receiver.fetch(in, Duration::SECOND*5));
+ BOOST_CHECK_EQUAL(out.getContent(), in.getContent());
+ control.unbind("my-exchange", "my-queue", "subject1");
+ control.bind("my-exchange", "my-queue", "subject2");
+
+ out.setContent("two");
+ sender.send(out);//should be dropped
+
+ out.setSubject("subject2");
+ out.setContent("three");
+ sender.send(out);//should not be dropped
+
+ BOOST_CHECK(receiver.fetch(in, Duration::SECOND*5));
+ BOOST_CHECK_EQUAL(out.getContent(), in.getContent());
+ BOOST_CHECK(!receiver.fetch(in, Duration::IMMEDIATE));
+ sender.close();
+ receiver.close();
+
+ control.deleteExchange("my-exchange");
+ messaging::Session other = fix.connection.createSession();
+ {
+ ScopedSuppressLogging sl;
+ BOOST_CHECK_THROW(other.createSender("my-exchange"), qpid::messaging::NotFound);
+ }
+ control.deleteQueue("my-queue");
+ other = fix.connection.createSession();
+ {
+ ScopedSuppressLogging sl;
+ BOOST_CHECK_THROW(other.createReceiver("my-queue"), qpid::messaging::NotFound);
+ }
+}
+
QPID_AUTO_TEST_SUITE_END()
}} // namespace qpid::tests
diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index 80c69ac386..57b344498e 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -244,7 +244,7 @@ QPID_AUTO_TEST_CASE(testBound){
exchange2.reset();
//unbind the queue from all exchanges it knows it has been bound to:
- queue->unbind(exchanges, queue);
+ queue->unbind(exchanges);
//ensure the remaining exchanges don't still have the queue bound to them:
FailOnDeliver deliverable;
diff --git a/cpp/src/tests/qpid-ctrl b/cpp/src/tests/qpid-ctrl
index 7b46c190fb..4246c57898 100755
--- a/cpp/src/tests/qpid-ctrl
+++ b/cpp/src/tests/qpid-ctrl
@@ -92,7 +92,10 @@ try:
arguments = {}
for a in args:
name, val = nameval(a)
- arguments[name] = val
+ if val[0] == '{' or val[0] == '[':
+ arguments[name] = eval(val)
+ else:
+ arguments[name] = val
content = {
"_object_id": {"_object_name": object_name},
"_method_name": method_name,
diff --git a/cpp/src/tests/sender.cpp b/cpp/src/tests/sender.cpp
index 9850e851da..063b5e87dc 100644
--- a/cpp/src/tests/sender.cpp
+++ b/cpp/src/tests/sender.cpp
@@ -120,7 +120,7 @@ void Sender::execute(AsyncSession& session, bool isRetry)
string data;
while (getline(std::cin, data)) {
message.setData(data);
- message.getHeaders().setInt("SN", ++sent);
+ //message.getHeaders().setInt("SN", ++sent);
string matchKey;
if (lvqMatchValues && getline(lvqMatchValues, matchKey)) {
message.getHeaders().setString(QueueOptions::strLVQMatchProperty, matchKey);
View Lorry Controller Status