diff options
Diffstat (limited to 'cpp/src')
| -rw-r--r-- | cpp/src/tests/MessagingFixture.h | 215 | ||||
| -rw-r--r-- | cpp/src/tests/MessagingSessionTests.cpp | 171 |
2 files changed, 216 insertions, 170 deletions
diff --git a/cpp/src/tests/MessagingFixture.h b/cpp/src/tests/MessagingFixture.h new file mode 100644 index 0000000000..5fb3fc56b2 --- /dev/null +++ b/cpp/src/tests/MessagingFixture.h @@ -0,0 +1,215 @@ +#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/Connection.h" +#include "qpid/messaging/Session.h" +#include "qpid/messaging/Sender.h" +#include "qpid/messaging/Receiver.h" +#include "qpid/messaging/Message.h" + +namespace qpid { +namespace tests { + +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(); + } + + ~BrokerAdmin() + { + session.close(); + connection.close(); + } +}; + +struct MessagingFixture : public BrokerFixture +{ + messaging::Connection connection; + messaging::Session session; + BrokerAdmin admin; + + MessagingFixture(Broker::Options opts = Broker::Options()) : + BrokerFixture(opts), + 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; + } + + 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); + } + } + +}; + +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; +} + + +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())); + } +} + +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()); + } +} + +}} // namespace qpid::tests + +#endif /*!TESTS_MESSAGINGFIXTURE_H*/ diff --git a/cpp/src/tests/MessagingSessionTests.cpp b/cpp/src/tests/MessagingSessionTests.cpp index e1b2a35213..2c51b9e093 100644 --- a/cpp/src/tests/MessagingSessionTests.cpp +++ b/cpp/src/tests/MessagingSessionTests.cpp @@ -18,9 +18,9 @@ * under the License. * */ +#include "MessagingFixture.h" #include "unit_test.h" #include "test_tools.h" -#include "BrokerFixture.h" #include "qpid/messaging/Address.h" #include "qpid/messaging/Connection.h" #include "qpid/messaging/Message.h" @@ -49,175 +49,6 @@ using namespace qpid; using qpid::broker::Broker; using qpid::framing::Uuid; -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(); - } - - ~BrokerAdmin() - { - session.close(); - connection.close(); - } -}; - -struct MessagingFixture : public BrokerFixture -{ - Connection connection; - Session session; - BrokerAdmin admin; - - MessagingFixture(Broker::Options opts = Broker::Options()) : - BrokerFixture(opts), - connection(open(broker->getPort(Broker::TCP_TRANSPORT))), - session(connection.createSession()), - admin(broker->getPort(Broker::TCP_TRANSPORT)) - { - } - - static Connection open(uint16_t port) - { - Connection connection((boost::format("amqp:tcp:localhost:%1%") % (port)).str()); - connection.open(); - return connection; - } - - 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*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); - } - } - -}; -std::vector<std::string> fetch(Receiver& receiver, int count, Duration timeout=Duration::SECOND*5) -{ - std::vector<std::string> data; - Message message; - for (int i = 0; i < count && receiver.fetch(message, timeout); i++) { - data.push_back(message.getContent()); - } - 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", Duration timeout=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()); - } -} QPID_AUTO_TEST_CASE(testSimpleSendReceive) { |