Creating a branch for experimenting with some ideas for JMS client.rajith_jms_client

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

1 files changed, 345 insertions, 0 deletions

diff --git a/qpid/cpp/src/tests/MessagingFixture.h b/qpid/cpp/src/tests/MessagingFixture.h
new file mode 100644
index 0000000000..2312a87e9d
--- /dev/null
+++ b/qpid/cpp/src/tests/MessagingFixture.h
@@ -0,0 +1,345 @@
+#ifndef TESTS_MESSAGINGFIXTURE_H
+#define TESTS_MESSAGINGFIXTURE_H
+
+/*
+ *
+ * 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 "BrokerFixture.h"
+#include "unit_test.h"
+#include "test_tools.h"
+#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;
+    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);
+    }
+
+    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();
+    }
+
+    void send(qpid::client::Message& message, const std::string& exchange=std::string())
+    {
+        session.messageTransfer(qpid::client::arg::destination=exchange, qpid::client::arg::content=message);
+    }
+
+    ~BrokerAdmin()
+    {
+        session.close();
+        connection.close();
+    }
+};
+
+struct MessagingFixture : public BrokerFixture
+{
+    messaging::Connection connection;
+    messaging::Session session;
+    BrokerAdmin admin;
+
+    MessagingFixture(Broker::Options opts = Broker::Options(), bool mgmtEnabled=false) :
+        BrokerFixture(opts, mgmtEnabled),
+        connection(open(broker->getPort(Broker::TCP_TRANSPORT))),
+        session(connection.createSession()),
+        admin(broker->getPort(Broker::TCP_TRANSPORT))
+    {
+    }
+
+    static messaging::Connection open(uint16_t port)
+    {
+        messaging::Connection connection(
+            (boost::format("amqp:tcp:localhost:%1%") % (port)).str());
+        connection.open();
+        return connection;
+    }
+
+    /** Open a connection to the broker. */
+    qpid::messaging::Connection newConnection()
+    {
+        qpid::messaging::Connection connection(
+            (boost::format("amqp:tcp:localhost:%1%") % (broker->getPort(qpid::broker::Broker::TCP_TRANSPORT))).str());
+        return connection;
+    }
+
+    void ping(const qpid::messaging::Address& address)
+    {
+        messaging::Receiver r = session.createReceiver(address);
+        messaging::Sender s = session.createSender(address);
+        messaging::Message out(framing::Uuid(true).str());
+        s.send(out);
+        messaging::Message in;
+        BOOST_CHECK(r.fetch(in, 5*messaging::Duration::SECOND));
+        BOOST_CHECK_EQUAL(out.getContent(), in.getContent());
+        r.close();
+        s.close();
+    }
+
+    ~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()
+    {
+        connection.close();
+        for (const_iterator i = queues.begin(); i != queues.end(); ++i) {
+            admin.deleteQueue(*i);
+        }
+    }
+
+};
+
+inline std::vector<std::string> fetch(messaging::Receiver& receiver, int count, messaging::Duration timeout=messaging::Duration::SECOND*5)
+{
+    std::vector<std::string> data;
+    messaging::Message message;
+    for (int i = 0; i < count && receiver.fetch(message, timeout); i++) {
+        data.push_back(message.getContent());
+    }
+    return data;
+}
+
+
+inline void send(messaging::Sender& sender, uint count = 1, uint start = 1,
+          const std::string& base = "Message")
+{
+    for (uint i = start; i < start + count; ++i) {
+        sender.send(messaging::Message((boost::format("%1%_%2%") % base % i).str()));
+    }
+}
+
+inline void receive(messaging::Receiver& receiver, uint count = 1, uint start = 1,
+             const std::string& base = "Message",
+             messaging::Duration timeout=messaging::Duration::SECOND*5)
+{
+    for (uint i = start; i < start + count; ++i) {
+        BOOST_CHECK_EQUAL(receiver.fetch(timeout).getContent(), (boost::format("%1%_%2%") % base % i).str());
+    }
+}
+
+
+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*/