diff options
author | Gordon Sim <gsim@apache.org> | 2007-05-08 17:00:00 +0000 |
---|---|---|
committer | Gordon Sim <gsim@apache.org> | 2007-05-08 17:00:00 +0000 |
commit | 6abe4fc19455e19069134c6b9161ef7eaee258b1 (patch) | |
tree | f8cec52653ef26a837aaf16578284b8828a128e8 | |
parent | 82a286a68212b128c13531d8188cc5fe947bd53d (diff) | |
download | qpid-python-6abe4fc19455e19069134c6b9161ef7eaee258b1.tar.gz |
Initial interop tests for c++.
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2@536253 13f79535-47bb-0310-9956-ffa450edef68
-rw-r--r-- | cpp/lib/broker/SessionHandlerImpl.cpp | 8 | ||||
-rw-r--r-- | cpp/tests/BasicP2PTest.h | 84 | ||||
-rw-r--r-- | cpp/tests/BasicPubSubTest.h | 144 | ||||
-rw-r--r-- | cpp/tests/Makefile.am | 3 | ||||
-rw-r--r-- | cpp/tests/SimpleTestCaseBase.h | 137 | ||||
-rw-r--r-- | cpp/tests/TestCase.h | 64 | ||||
-rw-r--r-- | cpp/tests/TestOptions.h | 69 | ||||
-rw-r--r-- | cpp/tests/interop_runner.cpp | 252 |
8 files changed, 758 insertions, 3 deletions
diff --git a/cpp/lib/broker/SessionHandlerImpl.cpp b/cpp/lib/broker/SessionHandlerImpl.cpp index 75fe11d6da..49767f4c71 100644 --- a/cpp/lib/broker/SessionHandlerImpl.cpp +++ b/cpp/lib/broker/SessionHandlerImpl.cpp @@ -233,8 +233,12 @@ void SessionHandlerImpl::ChannelHandlerImpl::open(u_int16_t channel, const strin } } -void SessionHandlerImpl::ChannelHandlerImpl::flow(u_int16_t /*channel*/, bool /*active*/){} -void SessionHandlerImpl::ChannelHandlerImpl::flowOk(u_int16_t /*channel*/, bool /*active*/){} +void SessionHandlerImpl::ChannelHandlerImpl::flow(u_int16_t /*channel*/, bool /*active*/){ + +} +void SessionHandlerImpl::ChannelHandlerImpl::flowOk(u_int16_t /*channel*/, bool /*active*/){ + +} void SessionHandlerImpl::ChannelHandlerImpl::close(u_int16_t channel, u_int16_t /*replyCode*/, const string& /*replyText*/, u_int16_t /*classId*/, u_int16_t /*methodId*/){ diff --git a/cpp/tests/BasicP2PTest.h b/cpp/tests/BasicP2PTest.h new file mode 100644 index 0000000000..f38ea2c387 --- /dev/null +++ b/cpp/tests/BasicP2PTest.h @@ -0,0 +1,84 @@ +#ifndef _BasicP2PTest_ +#define _BasicP2PTest_ +/* + * + * 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 <memory> +#include <sstream> + +#include <ClientChannel.h> +#include <ClientMessage.h> +#include <Connection.h> +#include <Exception.h> +#include <MessageListener.h> +#include "SimpleTestCaseBase.h" + + +namespace qpid { + +using namespace qpid::client; + +class BasicP2PTest : public SimpleTestCaseBase +{ + + class Receiver : public Worker, public MessageListener + { + const std::string queue; + std::string tag; + public: + Receiver(TestOptions& options, const std::string& _queue, const int _messages) : Worker(options, _messages), queue(_queue){} + + void init() + { + Queue q(queue, true); + channel.declareQueue(q); + channel.consume(q, tag, this); + channel.start(); + } + + void start(){ + } + + void received(Message&) + { + count++; + } + }; + +public: + void assign(const std::string& role, framing::FieldTable& params, TestOptions& options) + { + std::string queue = params.getString("P2P_QUEUE_AND_KEY_NAME"); + int messages = params.getInt("P2P_NUM_MESSAGES"); + if (role == "SENDER") { + worker = std::auto_ptr<Worker>(new Sender(options, Exchange::DEFAULT_EXCHANGE, queue, messages)); + } else if(role == "RECEIVER"){ + worker = std::auto_ptr<Worker>(new Receiver(options, queue, messages)); + } else { + throw Exception("unrecognised role"); + } + worker->init(); + } +}; + +} + +#endif diff --git a/cpp/tests/BasicPubSubTest.h b/cpp/tests/BasicPubSubTest.h new file mode 100644 index 0000000000..c86d85e81a --- /dev/null +++ b/cpp/tests/BasicPubSubTest.h @@ -0,0 +1,144 @@ +#ifndef _BasicPubSubTest_ +#define _BasicPubSubTest_ +/* + * + * 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 <memory> +#include <sstream> + +#include <ClientChannel.h> +#include <ClientMessage.h> +#include <Connection.h> +#include <Exception.h> +#include <MessageListener.h> +#include "SimpleTestCaseBase.h" +#include <boost/ptr_container/ptr_vector.hpp> +#include <boost/format.hpp> + + +namespace qpid { + +using namespace qpid::client; + +class BasicPubSubTest : public SimpleTestCaseBase +{ + + class Receiver : public Worker, public MessageListener + { + const Exchange& exchange; + const std::string queue; + const std::string key; + std::string tag; + public: + Receiver(TestOptions& options, const Exchange& _exchange, const std::string& _queue, const std::string& _key, const int _messages) + : Worker(options, _messages), exchange(_exchange), queue(_queue), key(_key){} + + void init() + { + Queue q(queue, true); + channel.declareQueue(q); + framing::FieldTable args; + channel.bind(exchange, q, key, args); + channel.consume(q, tag, this); + channel.start(); + } + + void start(){ + } + + void received(Message&) + { + count++; + } + }; + + class MultiReceiver : public Worker, public MessageListener + { + typedef boost::ptr_vector<Receiver> ReceiverList; + ReceiverList receivers; + + public: + MultiReceiver(TestOptions& options, const Exchange& exchange, const std::string& key, const int _messages, int receiverCount) + : Worker(options, _messages) + { + for (int i = 0; i != receiverCount; i++) { + std::string queue = (boost::format("%1%_%2%") % options.clientid % i).str(); + receivers.push_back(new Receiver(options, exchange, queue, key, _messages)); + } + } + + void init() + { + for (ReceiverList::size_type i = 0; i != receivers.size(); i++) { + receivers[i].init(); + } + } + + void start() + { + for (ReceiverList::size_type i = 0; i != receivers.size(); i++) { + receivers[i].start(); + } + } + + void received(Message& msg) + { + for (ReceiverList::size_type i = 0; i != receivers.size(); i++) { + receivers[i].received(msg); + } + } + + virtual int getCount() + { + count = 0; + for (ReceiverList::size_type i = 0; i != receivers.size(); i++) { + count += receivers[i].getCount(); + } + return count; + } + virtual void stop() + { + for (ReceiverList::size_type i = 0; i != receivers.size(); i++) { + receivers[i].stop(); + } + } + }; + +public: + void assign(const std::string& role, framing::FieldTable& params, TestOptions& options) + { + std::string key = params.getString("PUBSUB_KEY"); + int messages = params.getInt("PUBSUB_NUM_MESSAGES"); + int receivers = params.getInt("PUBSUB_NUM_RECEIVERS"); + if (role == "SENDER") { + worker = std::auto_ptr<Worker>(new Sender(options, Exchange::STANDARD_TOPIC_EXCHANGE, key, messages)); + } else if(role == "RECEIVER"){ + worker = std::auto_ptr<Worker>(new MultiReceiver(options, Exchange::STANDARD_TOPIC_EXCHANGE, key, messages, receivers)); + } else { + throw Exception("unrecognised role"); + } + worker->init(); + } +}; + +} + +#endif diff --git a/cpp/tests/Makefile.am b/cpp/tests/Makefile.am index 2e3bf289c5..6e5208aca7 100644 --- a/cpp/tests/Makefile.am +++ b/cpp/tests/Makefile.am @@ -23,7 +23,8 @@ client_tests = \ client_test \ echo_service \ topic_listener \ - topic_publisher + topic_publisher \ + interop_runner broker_tests = \ AccumulatedAckTest \ diff --git a/cpp/tests/SimpleTestCaseBase.h b/cpp/tests/SimpleTestCaseBase.h new file mode 100644 index 0000000000..7a2421af3b --- /dev/null +++ b/cpp/tests/SimpleTestCaseBase.h @@ -0,0 +1,137 @@ +#ifndef _SimpleTestCaseBase_ +#define _SimpleTestCaseBase_ +/* + * + * 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 <memory> +#include <sstream> + +#include <ClientChannel.h> +#include <ClientMessage.h> +#include <Connection.h> +#include <Exception.h> +#include <MessageListener.h> +#include "TestCase.h" + + +namespace qpid { + +using namespace qpid::client; + +class SimpleTestCaseBase : public TestCase +{ +protected: + class Worker + { + protected: + client::Connection connection; + client::Channel channel; + const int messages; + int count; + + public: + + Worker(TestOptions& options, const int _messages) : + connection(options.trace), messages(_messages), count(0) + { + connection.open(options.broker, options.port); + connection.openChannel(&channel); + } + + virtual ~Worker(){} + + virtual void stop() + { + channel.close(); + connection.close(); + } + + virtual int getCount() + { + return count; + } + + virtual void init() = 0; + virtual void start() = 0; + }; + + class Sender : public Worker + { + const Exchange& exchange; + const std::string key; + public: + Sender(TestOptions& options, + const Exchange& _exchange, + const std::string& _key, + const int _messages) + : Worker(options, _messages), exchange(_exchange), key(_key) {} + + void init() + { + channel.start(); + } + + void start(){ + Message msg; + while (count < messages) { + channel.publish(msg, exchange, key); + count++; + } + stop(); + } + }; + + std::auto_ptr<Worker> worker; + +public: + virtual void assign(const std::string& role, framing::FieldTable& params, TestOptions& options) = 0; + + virtual ~SimpleTestCaseBase() {} + + void start() + { + if (worker.get()) { + worker->start(); + } + } + + void stop() + { + if (worker.get()) { + worker->stop(); + } + } + + void report(client::Message& report) + { + if (worker.get()) { + report.getHeaders().setInt("MESSAGE_COUNT", worker->getCount()); + //add number of messages sent or received + std::stringstream reportstr; + reportstr << worker->getCount(); + report.setData(reportstr.str()); + } + } +}; + +} + +#endif diff --git a/cpp/tests/TestCase.h b/cpp/tests/TestCase.h new file mode 100644 index 0000000000..a88b76bc1d --- /dev/null +++ b/cpp/tests/TestCase.h @@ -0,0 +1,64 @@ +#ifndef _TestCase_ +#define _TestCase_ +/* + * + * 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 <ClientMessage.h> +#include "TestOptions.h" + + +namespace qpid { + +/** + * Interface to be implemented by test cases for use with the test + * runner. + */ +class TestCase +{ +public: + /** + * Directs the test case to act in a particular role. Some roles + * may be 'activated' at this stage others may require an explicit + * start request. + */ + virtual void assign(const std::string& role, framing::FieldTable& params, TestOptions& options) = 0; + /** + * Each test will be started on its own thread, which should block + * until the test completes (this may or may not require an + * explicit stop() request). + */ + virtual void start() = 0; + /** + * Requests that the test be stopped if still running. + */ + virtual void stop() = 0; + /** + * Allows the test to fill in details on the final report + * message. Will be called only after start has returned. + */ + virtual void report(client::Message& report) = 0; + + virtual ~TestCase() {} +}; + +} + +#endif diff --git a/cpp/tests/TestOptions.h b/cpp/tests/TestOptions.h new file mode 100644 index 0000000000..6cf6314bca --- /dev/null +++ b/cpp/tests/TestOptions.h @@ -0,0 +1,69 @@ +#ifndef _TestOptions_ +#define _TestOptions_ +/* + * + * 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 <CommonOptions.h> + +namespace qpid { + +struct TestOptions : public qpid::CommonOptions +{ + TestOptions() : desc("Options"), broker("localhost"), virtualhost(""), clientid("cpp"), help(false) + { + using namespace qpid::program_options; + using namespace boost::program_options; + CommonOptions::addTo(desc); + desc.add_options() + ("broker,b", optValue(broker, "HOSTNAME"), "the hostname to connect to") + ("virtualhost,v", optValue(virtualhost, "VIRTUAL_HOST"), "virtual host") + ("clientid,i", optValue(clientid, "ID"), "unique client identifier") + ("help,h", optValue(help), "print this usage statement"); + } + + void parse(int argc, char** argv) + { + using namespace boost::program_options; + try { + variables_map vm; + store(parse_command_line(argc, argv, desc), vm); + notify(vm); + } catch(const error& e) { + std::cerr << "Error: " << e.what() << std::endl + << "Specify '--help' for usage." << std::endl; + } + } + + void usage() + { + std::cout << desc << std::endl; + } + + boost::program_options::options_description desc; + std::string broker; + std::string virtualhost; + std::string clientid; + bool help; +}; + +} + +#endif diff --git a/cpp/tests/interop_runner.cpp b/cpp/tests/interop_runner.cpp new file mode 100644 index 0000000000..3dd87a0735 --- /dev/null +++ b/cpp/tests/interop_runner.cpp @@ -0,0 +1,252 @@ +/* + * + * 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 <CommonOptions.h> +#include <Connection.h> +#include <ClientChannel.h> +#include <ClientExchange.h> +#include <ClientQueue.h> +#include <Exception.h> +#include <MessageListener.h> +#include <QpidError.h> +#include <sys/Thread.h> +#include <sys/Time.h> +#include <iostream> +#include <memory> +#include "BasicP2PTest.h" +#include "BasicPubSubTest.h" +#include "TestCase.h" +#include <boost/ptr_container/ptr_map.hpp> + +/** + * Framework for interop tests. + * + * [see http://cwiki.apache.org/confluence/display/qpid/Interop+Testing+Specification for details]. + */ + +using namespace qpid::client; +using namespace qpid::sys; +using qpid::TestCase; +using qpid::TestOptions; +using qpid::framing::FieldTable; +using std::string; + +class DummyRun : public TestCase +{ +public: + DummyRun() {} + void assign(const std::string&, FieldTable&, TestOptions&) {} + void start() {} + void stop() {} + void report(qpid::client::Message&) {} +}; + +string parse_next_word(const string& input, const string& delims, string::size_type& position); + +/** + */ +class Listener : public MessageListener, private Runnable{ + typedef boost::ptr_map<std::string, TestCase> TestMap; + + Channel& channel; + TestOptions& options; + TestMap tests; + const string name; + const string topic; + TestMap::iterator test; + std::auto_ptr<Thread> runner; + string reportTo; + string reportCorrelator; + + void shutdown(); + bool invite(const string& name); + void run(); + + void sendResponse(Message& response, string replyTo); + void sendResponse(Message& response, Message& request); + void sendSimpleResponse(const string& type, Message& request); + void sendReport(); +public: + Listener(Channel& channel, TestOptions& options); + void received(Message& msg); + void bindAndConsume(); + void registerTest(std::string name, TestCase* test); +}; + +/** + */ +int main(int argc, char** argv){ + TestOptions options; + options.parse(argc, argv); + + if (options.help) { + options.usage(); + } else { + try{ + Connection connection(options.trace); + connection.open(options.broker, options.port, "guest", "guest", options.virtualhost); + + Channel channel; + connection.openChannel(&channel); + + Listener listener(channel, options); + listener.registerTest("TC1_DummyRun", new DummyRun()); + listener.registerTest("TC2_BasicP2P", new qpid::BasicP2PTest()); + listener.registerTest("TC3_BasicPubSub", new qpid::BasicPubSubTest()); + + listener.bindAndConsume(); + + channel.run(); + connection.close(); + } catch(qpid::Exception error) { + std::cout << error.what() << std::endl; + } + } +} + +Listener::Listener(Channel& _channel, TestOptions& _options) : channel(_channel), options(_options), name(options.clientid), topic("iop.control." + name) +{} + +void Listener::registerTest(std::string name, TestCase* test) +{ + tests.insert(name, test); +} + +void Listener::bindAndConsume() +{ + Queue control(name, true); + channel.declareQueue(control); + qpid::framing::FieldTable bindArgs; + //replace these separate binds with a wildcard once that is supported on java broker + channel.bind(Exchange::STANDARD_TOPIC_EXCHANGE, control, "iop.control", bindArgs); + channel.bind(Exchange::STANDARD_TOPIC_EXCHANGE, control, topic, bindArgs); + + std::string tag; + channel.consume(control, tag, this); +} + +void Listener::sendSimpleResponse(const string& type, Message& request) +{ + Message response; + response.getHeaders().setString("CONTROL_TYPE", type); + response.getHeaders().setString("CLIENT_NAME", name); + response.getHeaders().setString("CLIENT_PRIVATE_CONTROL_KEY", topic); + response.setCorrelationId(request.getCorrelationId()); + sendResponse(response, request); +} + +void Listener::sendResponse(Message& response, Message& request) +{ + sendResponse(response, request.getReplyTo()); +} + +void Listener::sendResponse(Message& response, string replyTo) +{ + //Exchange and routing key need to be extracted from the reply-to + //field. Format is assumed to be: + // + // <exchange type>://<exchange name>/<routing key>?<options> + // + //and all we need is the exchange name and routing key + // + if (replyTo.empty()) throw qpid::Exception("Reply address not set!"); + const string delims(":/?="); + + string::size_type start = replyTo.find(':');//skip exchange type + string exchange = parse_next_word(replyTo, delims, start); + string routingKey = parse_next_word(replyTo, delims, start); + channel.publish(response, exchange, routingKey); +} + +void Listener::received(Message& message) +{ + std::string type(message.getHeaders().getString("CONTROL_TYPE")); + + if (type == "INVITE") { + std::string name(message.getHeaders().getString("TEST_NAME")); + if (name.empty() || invite(name)) { + sendSimpleResponse("ENLIST", message); + } else { + std::cout << "Can't take part in '" << name << "'" << std::endl; + } + } else if (type == "ASSIGN_ROLE") { + test->assign(message.getHeaders().getString("ROLE"), message.getHeaders(), options); + sendSimpleResponse("ACCEPT_ROLE", message); + } else if (type == "START") { + reportTo = message.getReplyTo(); + reportCorrelator = message.getCorrelationId(); + runner = std::auto_ptr<Thread>(new Thread(this)); + } else if (type == "STATUS_REQUEST") { + reportTo = message.getReplyTo(); + reportCorrelator = message.getCorrelationId(); + test->stop(); + sendReport(); + } else if (type == "TERMINATE") { + if (test != tests.end()) test->stop(); + shutdown(); + } else { + std::cerr <<"ERROR!: Received unknown control message: " << type << std::endl; + shutdown(); + } +} + +void Listener::shutdown() +{ + channel.close(); +} + +bool Listener::invite(const string& name) +{ + test = tests.find(name); + return test != tests.end(); +} + +void Listener::run() +{ + //NB: this method will be called in its own thread + //start test and when start returns... + test->start(); + sendReport(); +} + +void Listener::sendReport() +{ + Message report; + report.getHeaders().setString("CONTROL_TYPE", "REPORT"); + test->report(report); + report.setCorrelationId(reportCorrelator); + sendResponse(report, reportTo); +} + +string parse_next_word(const string& input, const string& delims, string::size_type& position) +{ + string::size_type start = input.find_first_not_of(delims, position); + if (start == string::npos) { + return ""; + } else { + string::size_type end = input.find_first_of(delims, start); + if (end == string::npos) { + end = input.length(); + } + position = end; + return input.substr(start, end - start); + } +} |