diff options
author | Alan Conway <aconway@apache.org> | 2008-05-23 21:23:07 +0000 |
---|---|---|
committer | Alan Conway <aconway@apache.org> | 2008-05-23 21:23:07 +0000 |
commit | 3b4d08876f63637cd9ffb28988eb2ec9a9a7f30e (patch) | |
tree | 11206920074aef9a5b9d794b6ed550f72d3a198d /cpp/src/tests | |
parent | 52833097fb1737316c76822bf7e6dda31dec3433 (diff) | |
download | qpid-python-3b4d08876f63637cd9ffb28988eb2ec9a9a7f30e.tar.gz |
Delete obsolete Channel class.
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@659663 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'cpp/src/tests')
-rw-r--r-- | cpp/src/tests/ClientChannelTest.cpp | 220 | ||||
-rw-r--r-- | cpp/src/tests/Makefile.am | 35 | ||||
-rw-r--r-- | cpp/src/tests/echo_service.cpp | 229 |
3 files changed, 18 insertions, 466 deletions
diff --git a/cpp/src/tests/ClientChannelTest.cpp b/cpp/src/tests/ClientChannelTest.cpp deleted file mode 100644 index 605d5e4885..0000000000 --- a/cpp/src/tests/ClientChannelTest.cpp +++ /dev/null @@ -1,220 +0,0 @@ -/* - * - * 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 <vector> -#include "qpid_test_plugin.h" -#include "BrokerFixture.h" -#include "qpid/client/Channel.h" -#include "qpid/client/Message.h" -#include "qpid/client/Queue.h" -#include "qpid/client/Exchange.h" -#include "qpid/client/MessageListener.h" -#include "qpid/client/BasicMessageChannel.h" -#include "qpid/client/MessageMessageChannel.h" - -using namespace std; -using namespace boost; -using namespace qpid::client; -using namespace qpid::sys; -using namespace qpid::framing; - -/// Small frame size so we can create fragmented messages. -const size_t FRAME_MAX = 256; - - -/** - * Test base for client API using an in-process broker. - * The test base defines the tests methods, derived classes - * instantiate the channel in Basic or Message mode. - */ -class ChannelTestBase : public CppUnit::TestCase, public SessionFixture -{ - struct Listener: public qpid::client::MessageListener { - vector<Message> messages; - Monitor monitor; - void received(Message& msg) { - Mutex::ScopedLock l(monitor); - messages.push_back(msg); - monitor.notifyAll(); - } - }; - - const std::string qname; - const std::string data; - Queue queue; - Exchange exchange; - Listener listener; - - protected: - boost::scoped_ptr<Channel> channel; - - public: - - ChannelTestBase() - : qname("testq"), data("hello"), - queue(qname, true), exchange("", Exchange::DIRECT_EXCHANGE) - {} - - void setUp() { - CPPUNIT_ASSERT(channel); - connection.openChannel(*channel); - CPPUNIT_ASSERT(channel->getId() != 0); - channel->declareQueue(queue); - } - - void testPublishGet() { - Message pubMsg(data); - pubMsg.getHeaders().setString("hello", "world"); - channel->publish(pubMsg, exchange, qname); - Message getMsg; - CPPUNIT_ASSERT(channel->get(getMsg, queue)); - CPPUNIT_ASSERT_EQUAL(data, getMsg.getData()); - CPPUNIT_ASSERT_EQUAL(string("world"), - getMsg.getHeaders().getString("hello")); - CPPUNIT_ASSERT(!channel->get(getMsg, queue)); // Empty queue - } - - void testGetNoContent() { - Message pubMsg; - pubMsg.getHeaders().setString("hello", "world"); - channel->publish(pubMsg, exchange, qname); - Message getMsg; - CPPUNIT_ASSERT(channel->get(getMsg, queue)); - CPPUNIT_ASSERT(getMsg.getData().empty()); - CPPUNIT_ASSERT_EQUAL(string("world"), - getMsg.getHeaders().getString("hello")); - } - - void testConsumeCancel() { - string tag; // Broker assigned - channel->consume(queue, tag, &listener); - channel->start(); - CPPUNIT_ASSERT_EQUAL(size_t(0), listener.messages.size()); - channel->publish(Message("a"), exchange, qname); - { - Mutex::ScopedLock l(listener.monitor); - Time deadline(now() + 1*TIME_SEC); - while (listener.messages.size() != 1) { - CPPUNIT_ASSERT(listener.monitor.wait(deadline)); - } - } - CPPUNIT_ASSERT_EQUAL(size_t(1), listener.messages.size()); - CPPUNIT_ASSERT_EQUAL(string("a"), listener.messages[0].getData()); - - channel->publish(Message("b"), exchange, qname); - channel->publish(Message("c"), exchange, qname); - { - Mutex::ScopedLock l(listener.monitor); - while (listener.messages.size() != 3) { - CPPUNIT_ASSERT(listener.monitor.wait(1*TIME_SEC)); - } - } - CPPUNIT_ASSERT_EQUAL(size_t(3), listener.messages.size()); - CPPUNIT_ASSERT_EQUAL(string("b"), listener.messages[1].getData()); - CPPUNIT_ASSERT_EQUAL(string("c"), listener.messages[2].getData()); - - channel->cancel(tag); - channel->publish(Message("d"), exchange, qname); - CPPUNIT_ASSERT_EQUAL(size_t(3), listener.messages.size()); - { - Mutex::ScopedLock l(listener.monitor); - CPPUNIT_ASSERT(!listener.monitor.wait(TIME_SEC/2)); - } - Message msg; - CPPUNIT_ASSERT(channel->get(msg, queue)); - CPPUNIT_ASSERT_EQUAL(string("d"), msg.getData()); - } - - // Consume already-published messages - void testConsumePublished() { - Message pubMsg("x"); - pubMsg.getHeaders().setString("y", "z"); - channel->publish(pubMsg, exchange, qname); - string tag; - channel->consume(queue, tag, &listener); - CPPUNIT_ASSERT_EQUAL(size_t(0), listener.messages.size()); - channel->start(); - { - Mutex::ScopedLock l(listener.monitor); - while (listener.messages.size() != 1) - CPPUNIT_ASSERT(listener.monitor.wait(1*TIME_SEC)); - } - CPPUNIT_ASSERT_EQUAL(string("x"), listener.messages[0].getData()); - CPPUNIT_ASSERT_EQUAL(string("z"), - listener.messages[0].getHeaders().getString("y")); - } - - void testGetFragmentedMessage() { - string longStr(FRAME_MAX*2, 'x'); // Longer than max frame size. - channel->publish(Message(longStr), exchange, qname); - Message getMsg; - CPPUNIT_ASSERT(channel->get(getMsg, queue)); - } - - void testConsumeFragmentedMessage() { - string xx(FRAME_MAX*2, 'x'); - channel->publish(Message(xx), exchange, qname); - channel->start(); - string tag; - channel->consume(queue, tag, &listener); - string yy(FRAME_MAX*2, 'y'); - channel->publish(Message(yy), exchange, qname); - { - Mutex::ScopedLock l(listener.monitor); - while (listener.messages.size() != 2) - CPPUNIT_ASSERT(listener.monitor.wait(1*TIME_SEC)); - } - CPPUNIT_ASSERT_EQUAL(xx, listener.messages[0].getData()); - CPPUNIT_ASSERT_EQUAL(yy, listener.messages[1].getData()); - } -}; - -class BasicChannelTest : public ChannelTestBase { - CPPUNIT_TEST_SUITE(BasicChannelTest); - CPPUNIT_TEST(testPublishGet); - CPPUNIT_TEST(testGetNoContent); - CPPUNIT_TEST(testConsumeCancel); - CPPUNIT_TEST(testConsumePublished); - CPPUNIT_TEST(testGetFragmentedMessage); - CPPUNIT_TEST(testConsumeFragmentedMessage); - CPPUNIT_TEST_SUITE_END(); - - public: - BasicChannelTest(){ - channel.reset(new Channel(false, 500, Channel::AMQP_08)); - } -}; - -class MessageChannelTest : public ChannelTestBase { - CPPUNIT_TEST_SUITE(MessageChannelTest); - CPPUNIT_TEST(testPublishGet); - CPPUNIT_TEST(testGetNoContent); - CPPUNIT_TEST(testGetFragmentedMessage); - CPPUNIT_TEST_SUITE_END(); - public: - MessageChannelTest() { - channel.reset(new Channel(false, 500, Channel::AMQP_09)); - } -}; - -// Make this test suite a plugin. -CPPUNIT_PLUGIN_IMPLEMENT(); -CPPUNIT_TEST_SUITE_REGISTRATION(BasicChannelTest); -CPPUNIT_TEST_SUITE_REGISTRATION(MessageChannelTest); diff --git a/cpp/src/tests/Makefile.am b/cpp/src/tests/Makefile.am index 4a1a8d9a66..6c68a9d648 100644 --- a/cpp/src/tests/Makefile.am +++ b/cpp/src/tests/Makefile.am @@ -102,9 +102,6 @@ broker_unit_tests = \ TxPublishTest \ MessageBuilderTest -#client_unit_tests = \ -# ClientChannelTest - framing_unit_tests = \ FramingTest \ HeaderTest \ @@ -121,10 +118,9 @@ unit_tests = \ testprogs= \ client_test \ topic_listener \ - topic_publisher -# echo_service + topic_publisher -check_PROGRAMS += $(testprogs) interop_runner publish consume +check_PROGRAMS += $(testprogs) publish consume TESTS_ENVIRONMENT = VALGRIND=$(VALGRIND) srcdir=$(srcdir) QPID_DATA_DIR= $(srcdir)/run_test @@ -172,17 +168,22 @@ gen.mk: Makefile.am CLEANFILES+=valgrind.out *.log *.vglog dummy_test $(unit_wrappers) MAINTAINERCLEANFILES=gen.mk -interop_runner_SOURCES = \ - interop_runner.cpp \ - SimpleTestCaseBase.cpp \ - BasicP2PTest.cpp \ - BasicPubSubTest.cpp \ - SimpleTestCaseBase.h \ - BasicP2PTest.h \ - BasicPubSubTest.h \ - TestCase.h \ - TestOptions.h -interop_runner_LDADD = $(lib_client) $(lib_common) $(extra_libs) +# FIXME aconway 2008-05-23: Disabled interop_runner because it uses +# the obsolete Channel class. Convert to Session and re-enable. +# +# check_PROGRAMS += interop_runner + +# interop_runner_SOURCES = \ +# interop_runner.cpp \ +# SimpleTestCaseBase.cpp \ +# BasicP2PTest.cpp \ +# BasicPubSubTest.cpp \ +# SimpleTestCaseBase.h \ +# BasicP2PTest.h \ +# BasicPubSubTest.h \ +# TestCase.h \ +# TestOptions.h +# interop_runner_LDADD = $(lib_client) $(lib_common) $(extra_libs) publish_SOURCES = publish.cpp publish_LDADD = $(lib_client) $(lib_common) $(extra_libs) diff --git a/cpp/src/tests/echo_service.cpp b/cpp/src/tests/echo_service.cpp deleted file mode 100644 index c3569d5fd4..0000000000 --- a/cpp/src/tests/echo_service.cpp +++ /dev/null @@ -1,229 +0,0 @@ -/* - * - * 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. - * - */ - -/** - * This class provides an example of using AMQP for a request-response - * style system. 'Requests' are messages sent to a well known - * destination. A 'service' process consumes these message and - * responds by echoing the message back to the sender on a - * sender-specified private queue. - */ - -#include "qpid/client/Channel.h" -#include "qpid/client/Connection.h" -#include "qpid/client/Exchange.h" -#include "qpid/client/MessageListener.h" -#include "qpid/client/Queue.h" -#include "qpid/sys/Time.h" -#include <iostream> -#include <sstream> - -using namespace qpid::client; -using namespace qpid::sys; -using std::string; - - -/** - * A message listener implementation representing the 'service', this - * will 'echo' any requests received. - */ -class EchoServer : public MessageListener{ - Channel* const channel; -public: - EchoServer(Channel* channel); - virtual void received(Message& msg); -}; - -/** - * A message listener implementation that merely prints received - * messages to the console. Used to report on 'echo' responses. - */ -class LoggingListener : public MessageListener{ -public: - virtual void received(Message& msg); -}; - -/** - * A utility class that manages the command line options needed to run - * the example confirgurably. - */ -class Args{ - string host; - int port; - bool trace; - bool help; - bool client; -public: - inline Args() : host("localhost"), port(5672), trace(false), help(false), client(false){} - void parse(int argc, char** argv); - void usage(); - - inline const string& getHost() const { return host;} - inline int getPort() const { return port; } - inline bool getTrace() const { return trace; } - inline bool getHelp() const { return help; } - inline bool getClient() const { return client; } -}; - -/** - * The main test path. There are two basic modes: 'client' and - * 'service'. First one or more services are started, then one or more - * clients are started and messages can be sent. - */ -int main(int argc, char** argv){ - const std::string echo_service("echo_service"); - Args args; - args.parse(argc, argv); - if (args.getHelp()) { - args.usage(); - } else if (args.getClient()) { - //we have been started in 'client' mode, i.e. we will send an - //echo requests and print responses received. - try { - //Create connection & open a channel - Connection connection(args.getTrace()); - connection.open(args.getHost(), args.getPort()); - Channel channel; - connection.openChannel(channel); - - //Setup: declare the private 'response' queue and bind it - //to the direct exchange by its name which will be - //generated by the server - Queue response; - channel.declareQueue(response); - qpid::framing::FieldTable emptyArgs; - channel.bind(Exchange::STANDARD_DIRECT_EXCHANGE, response, response.getName(), emptyArgs); - - //Consume from the response queue, logging all echoed message to console: - LoggingListener listener; - std::string tag; - channel.consume(response, tag, &listener); - - //Process incoming requests on a new thread - channel.start(); - - //get messages from console and send them: - std::string text; - std::cout << "Enter text to send:" << std::endl; - while (std::getline(std::cin, text)) { - std::cout << "Sending " << text << " to echo server." << std::endl; - Message msg; - msg.getHeaders().setString("RESPONSE_QUEUE", response.getName()); - msg.setData(text); - channel.publish(msg, Exchange::STANDARD_DIRECT_EXCHANGE, echo_service); - - std::cout << "Enter text to send:" << std::endl; - } - - connection.close(); - } catch(std::exception& error) { - std::cout << error.what() << std::endl; - } - } else { - // we are in 'service' mode, i.e. we will consume messages - // from the request queue and echo each request back to the - // senders own private response queue. - try { - //Create connection & open a channel - Connection connection(args.getTrace()); - connection.open(args.getHost(), args.getPort()); - Channel channel; - connection.openChannel(channel); - - //Setup: declare the 'request' queue and bind it to the direct exchange with a 'well known' name - Queue request("request"); - channel.declareQueue(request); - qpid::framing::FieldTable emptyArgs; - channel.bind(Exchange::STANDARD_DIRECT_EXCHANGE, request, echo_service, emptyArgs); - - //Consume from the request queue, echoing back all messages received to the client that sent them - EchoServer server(&channel); - std::string tag = "server_tag"; - channel.consume(request, tag, &server); - - //Process incoming requests on the main thread - channel.run(); - - connection.close(); - } catch(std::exception& error) { - std::cout << error.what() << std::endl; - } - } -} - -EchoServer::EchoServer(Channel* _channel) : channel(_channel){} - -void EchoServer::received(Message& message) -{ - //get name of response queues binding to the default direct exchange: - const std::string name = message.getHeaders().getString("RESPONSE_QUEUE"); - - if (name.empty()) { - std::cout << "Cannot echo " << message.getData() << ", no response queue specified." << std::endl; - } else { - //print message to console: - std::cout << "Echoing " << message.getData() << " back to " << name << std::endl; - - //'echo' the message back: - channel->publish(message, Exchange::STANDARD_DIRECT_EXCHANGE, name); - } -} - -void LoggingListener::received(Message& message) -{ - //print message to console: - std::cout << "Received echo: " << message.getData() << std::endl; -} - - -void Args::parse(int argc, char** argv){ - for(int i = 1; i < argc; i++){ - string name(argv[i]); - if("-help" == name){ - help = true; - break; - }else if("-host" == name){ - host = argv[++i]; - }else if("-port" == name){ - port = atoi(argv[++i]); - }else if("-trace" == name){ - trace = true; - }else if("-client" == name){ - client = true; - }else{ - std::cout << "Warning: unrecognised option " << name << std::endl; - } - } -} - -void Args::usage(){ - std::cout << "Options:" << std::endl; - std::cout << " -help" << std::endl; - std::cout << " Prints this usage message" << std::endl; - std::cout << " -host <host>" << std::endl; - std::cout << " Specifies host to connect to (default is localhost)" << std::endl; - std::cout << " -port <port>" << std::endl; - std::cout << " Specifies port to conect to (default is 5762)" << std::endl; - std::cout << " -trace" << std::endl; - std::cout << " Indicates that the frames sent and received should be logged" << std::endl; - std::cout << " -client" << std::endl; - std::cout << " Run as a client (else will run as a server)" << std::endl; -} |